SuperRealCo commited on
Commit
2d0fb03
·
verified ·
1 Parent(s): e56adfa

Delete folder_paths.py

Browse files
Files changed (1) hide show
  1. folder_paths.py +0 -425
folder_paths.py DELETED
@@ -1,425 +0,0 @@
1
- from __future__ import annotations
2
-
3
- import os
4
- import time
5
- import mimetypes
6
- import logging
7
- from typing import Literal, List
8
- from collections.abc import Collection
9
-
10
- from comfy.cli_args import args
11
-
12
- supported_pt_extensions: set[str] = {'.ckpt', '.pt', '.pt2', '.bin', '.pth', '.safetensors', '.pkl', '.sft'}
13
-
14
- folder_names_and_paths: dict[str, tuple[list[str], set[str]]] = {}
15
-
16
- # --base-directory - Resets all default paths configured in folder_paths with a new base path
17
- if args.base_directory:
18
- base_path = os.path.abspath(args.base_directory)
19
- else:
20
- base_path = os.path.dirname(os.path.realpath(__file__))
21
-
22
- models_dir = os.path.join(base_path, "models")
23
- folder_names_and_paths["checkpoints"] = ([os.path.join(models_dir, "checkpoints")], supported_pt_extensions)
24
- folder_names_and_paths["configs"] = ([os.path.join(models_dir, "configs")], [".yaml"])
25
-
26
- folder_names_and_paths["loras"] = ([os.path.join(models_dir, "loras")], supported_pt_extensions)
27
- folder_names_and_paths["vae"] = ([os.path.join(models_dir, "vae")], supported_pt_extensions)
28
- folder_names_and_paths["text_encoders"] = ([os.path.join(models_dir, "text_encoders"), os.path.join(models_dir, "clip")], supported_pt_extensions)
29
- folder_names_and_paths["diffusion_models"] = ([os.path.join(models_dir, "unet"), os.path.join(models_dir, "diffusion_models")], supported_pt_extensions)
30
- folder_names_and_paths["clip_vision"] = ([os.path.join(models_dir, "clip_vision")], supported_pt_extensions)
31
- folder_names_and_paths["style_models"] = ([os.path.join(models_dir, "style_models")], supported_pt_extensions)
32
- folder_names_and_paths["embeddings"] = ([os.path.join(models_dir, "embeddings")], supported_pt_extensions)
33
- folder_names_and_paths["diffusers"] = ([os.path.join(models_dir, "diffusers")], ["folder"])
34
- folder_names_and_paths["vae_approx"] = ([os.path.join(models_dir, "vae_approx")], supported_pt_extensions)
35
-
36
- folder_names_and_paths["controlnet"] = ([os.path.join(models_dir, "controlnet"), os.path.join(models_dir, "t2i_adapter")], supported_pt_extensions)
37
- folder_names_and_paths["gligen"] = ([os.path.join(models_dir, "gligen")], supported_pt_extensions)
38
-
39
- folder_names_and_paths["upscale_models"] = ([os.path.join(models_dir, "upscale_models")], supported_pt_extensions)
40
-
41
- folder_names_and_paths["custom_nodes"] = ([os.path.join(base_path, "custom_nodes")], set())
42
-
43
- folder_names_and_paths["hypernetworks"] = ([os.path.join(models_dir, "hypernetworks")], supported_pt_extensions)
44
-
45
- folder_names_and_paths["photomaker"] = ([os.path.join(models_dir, "photomaker")], supported_pt_extensions)
46
-
47
- folder_names_and_paths["classifiers"] = ([os.path.join(models_dir, "classifiers")], {""})
48
-
49
- output_directory = os.path.join(base_path, "output")
50
- temp_directory = os.path.join(base_path, "temp")
51
- input_directory = os.path.join(base_path, "input")
52
- user_directory = os.path.join(base_path, "user")
53
-
54
- filename_list_cache: dict[str, tuple[list[str], dict[str, float], float]] = {}
55
-
56
- class CacheHelper:
57
- """
58
- Helper class for managing file list cache data.
59
- """
60
- def __init__(self):
61
- self.cache: dict[str, tuple[list[str], dict[str, float], float]] = {}
62
- self.active = False
63
-
64
- def get(self, key: str, default=None) -> tuple[list[str], dict[str, float], float]:
65
- if not self.active:
66
- return default
67
- return self.cache.get(key, default)
68
-
69
- def set(self, key: str, value: tuple[list[str], dict[str, float], float]) -> None:
70
- if self.active:
71
- self.cache[key] = value
72
-
73
- def clear(self):
74
- self.cache.clear()
75
-
76
- def __enter__(self):
77
- self.active = True
78
- return self
79
-
80
- def __exit__(self, exc_type, exc_value, traceback):
81
- self.active = False
82
- self.clear()
83
-
84
- cache_helper = CacheHelper()
85
-
86
- extension_mimetypes_cache = {
87
- "webp" : "image",
88
- "fbx" : "model",
89
- }
90
-
91
- def map_legacy(folder_name: str) -> str:
92
- legacy = {"unet": "diffusion_models",
93
- "clip": "text_encoders"}
94
- return legacy.get(folder_name, folder_name)
95
-
96
- if not os.path.exists(input_directory):
97
- try:
98
- os.makedirs(input_directory)
99
- except:
100
- logging.error("Failed to create input directory")
101
-
102
- def set_output_directory(output_dir: str) -> None:
103
- global output_directory
104
- output_directory = output_dir
105
-
106
- def set_temp_directory(temp_dir: str) -> None:
107
- global temp_directory
108
- temp_directory = temp_dir
109
-
110
- def set_input_directory(input_dir: str) -> None:
111
- global input_directory
112
- input_directory = input_dir
113
-
114
- def get_output_directory() -> str:
115
- global output_directory
116
- return output_directory
117
-
118
- def get_temp_directory() -> str:
119
- global temp_directory
120
- return temp_directory
121
-
122
- def get_input_directory() -> str:
123
- global input_directory
124
- return input_directory
125
-
126
- def get_user_directory() -> str:
127
- return user_directory
128
-
129
- def set_user_directory(user_dir: str) -> None:
130
- global user_directory
131
- user_directory = user_dir
132
-
133
-
134
- #NOTE: used in http server so don't put folders that should not be accessed remotely
135
- def get_directory_by_type(type_name: str) -> str | None:
136
- if type_name == "output":
137
- return get_output_directory()
138
- if type_name == "temp":
139
- return get_temp_directory()
140
- if type_name == "input":
141
- return get_input_directory()
142
- return None
143
-
144
- def filter_files_content_types(files: list[str], content_types: List[Literal["image", "video", "audio", "model"]]) -> list[str]:
145
- """
146
- Example:
147
- files = os.listdir(folder_paths.get_input_directory())
148
- videos = filter_files_content_types(files, ["video"])
149
-
150
- Note:
151
- - 'model' in MIME context refers to 3D models, not files containing trained weights and parameters
152
- """
153
- global extension_mimetypes_cache
154
- result = []
155
- for file in files:
156
- extension = file.split('.')[-1]
157
- if extension not in extension_mimetypes_cache:
158
- mime_type, _ = mimetypes.guess_type(file, strict=False)
159
- if not mime_type:
160
- continue
161
- content_type = mime_type.split('/')[0]
162
- extension_mimetypes_cache[extension] = content_type
163
- else:
164
- content_type = extension_mimetypes_cache[extension]
165
-
166
- if content_type in content_types:
167
- result.append(file)
168
- return result
169
-
170
- # determine base_dir rely on annotation if name is 'filename.ext [annotation]' format
171
- # otherwise use default_path as base_dir
172
- def annotated_filepath(name: str) -> tuple[str, str | None]:
173
- if name.endswith("[output]"):
174
- base_dir = get_output_directory()
175
- name = name[:-9]
176
- elif name.endswith("[input]"):
177
- base_dir = get_input_directory()
178
- name = name[:-8]
179
- elif name.endswith("[temp]"):
180
- base_dir = get_temp_directory()
181
- name = name[:-7]
182
- else:
183
- return name, None
184
-
185
- return name, base_dir
186
-
187
-
188
- def get_annotated_filepath(name: str, default_dir: str | None=None) -> str:
189
- name, base_dir = annotated_filepath(name)
190
-
191
- if base_dir is None:
192
- if default_dir is not None:
193
- base_dir = default_dir
194
- else:
195
- base_dir = get_input_directory() # fallback path
196
-
197
- return os.path.join(base_dir, name)
198
-
199
-
200
- def exists_annotated_filepath(name) -> bool:
201
- name, base_dir = annotated_filepath(name)
202
-
203
- if base_dir is None:
204
- base_dir = get_input_directory() # fallback path
205
-
206
- filepath = os.path.join(base_dir, name)
207
- return os.path.exists(filepath)
208
-
209
-
210
- def add_model_folder_path(folder_name: str, full_folder_path: str, is_default: bool = False) -> None:
211
- global folder_names_and_paths
212
- folder_name = map_legacy(folder_name)
213
- if folder_name in folder_names_and_paths:
214
- paths, _exts = folder_names_and_paths[folder_name]
215
- if full_folder_path in paths:
216
- if is_default and paths[0] != full_folder_path:
217
- # If the path to the folder is not the first in the list, move it to the beginning.
218
- paths.remove(full_folder_path)
219
- paths.insert(0, full_folder_path)
220
- else:
221
- if is_default:
222
- paths.insert(0, full_folder_path)
223
- else:
224
- paths.append(full_folder_path)
225
- else:
226
- folder_names_and_paths[folder_name] = ([full_folder_path], set())
227
-
228
- def get_folder_paths(folder_name: str) -> list[str]:
229
- folder_name = map_legacy(folder_name)
230
- return folder_names_and_paths[folder_name][0][:]
231
-
232
- def recursive_search(directory: str, excluded_dir_names: list[str] | None=None) -> tuple[list[str], dict[str, float]]:
233
- if not os.path.isdir(directory):
234
- return [], {}
235
-
236
- if excluded_dir_names is None:
237
- excluded_dir_names = []
238
-
239
- result = []
240
- dirs = {}
241
-
242
- # Attempt to add the initial directory to dirs with error handling
243
- try:
244
- dirs[directory] = os.path.getmtime(directory)
245
- except FileNotFoundError:
246
- logging.warning(f"Warning: Unable to access {directory}. Skipping this path.")
247
-
248
- logging.debug("recursive file list on directory {}".format(directory))
249
- dirpath: str
250
- subdirs: list[str]
251
- filenames: list[str]
252
-
253
- for dirpath, subdirs, filenames in os.walk(directory, followlinks=True, topdown=True):
254
- subdirs[:] = [d for d in subdirs if d not in excluded_dir_names]
255
- for file_name in filenames:
256
- try:
257
- relative_path = os.path.relpath(os.path.join(dirpath, file_name), directory)
258
- result.append(relative_path)
259
- except:
260
- logging.warning(f"Warning: Unable to access {file_name}. Skipping this file.")
261
- continue
262
-
263
- for d in subdirs:
264
- path: str = os.path.join(dirpath, d)
265
- try:
266
- dirs[path] = os.path.getmtime(path)
267
- except FileNotFoundError:
268
- logging.warning(f"Warning: Unable to access {path}. Skipping this path.")
269
- continue
270
- logging.debug("found {} files".format(len(result)))
271
- return result, dirs
272
-
273
- def filter_files_extensions(files: Collection[str], extensions: Collection[str]) -> list[str]:
274
- return sorted(list(filter(lambda a: os.path.splitext(a)[-1].lower() in extensions or len(extensions) == 0, files)))
275
-
276
-
277
-
278
- def get_full_path(folder_name: str, filename: str) -> str | None:
279
- """
280
- Get the full path of a file in a folder, has to be a file
281
- """
282
- global folder_names_and_paths
283
- folder_name = map_legacy(folder_name)
284
- if folder_name not in folder_names_and_paths:
285
- return None
286
- folders = folder_names_and_paths[folder_name]
287
- filename = os.path.relpath(os.path.join("/", filename), "/")
288
- for x in folders[0]:
289
- full_path = os.path.join(x, filename)
290
- if os.path.isfile(full_path):
291
- return full_path
292
- elif os.path.islink(full_path):
293
- logging.warning("WARNING path {} exists but doesn't link anywhere, skipping.".format(full_path))
294
-
295
- return None
296
-
297
-
298
- def get_full_path_or_raise(folder_name: str, filename: str) -> str:
299
- """
300
- Get the full path of a file in a folder, has to be a file
301
- """
302
- full_path = get_full_path(folder_name, filename)
303
- if full_path is None:
304
- raise FileNotFoundError(f"Model in folder '{folder_name}' with filename '{filename}' not found.")
305
- return full_path
306
-
307
-
308
- def get_filename_list_(folder_name: str) -> tuple[list[str], dict[str, float], float]:
309
- folder_name = map_legacy(folder_name)
310
- global folder_names_and_paths
311
- output_list = set()
312
- folders = folder_names_and_paths[folder_name]
313
- output_folders = {}
314
- for x in folders[0]:
315
- files, folders_all = recursive_search(x, excluded_dir_names=[".git"])
316
- output_list.update(filter_files_extensions(files, folders[1]))
317
- output_folders = {**output_folders, **folders_all}
318
-
319
- return sorted(list(output_list)), output_folders, time.perf_counter()
320
-
321
- def cached_filename_list_(folder_name: str) -> tuple[list[str], dict[str, float], float] | None:
322
- strong_cache = cache_helper.get(folder_name)
323
- if strong_cache is not None:
324
- return strong_cache
325
-
326
- global filename_list_cache
327
- global folder_names_and_paths
328
- folder_name = map_legacy(folder_name)
329
- if folder_name not in filename_list_cache:
330
- return None
331
- out = filename_list_cache[folder_name]
332
-
333
- for x in out[1]:
334
- time_modified = out[1][x]
335
- folder = x
336
- if os.path.getmtime(folder) != time_modified:
337
- return None
338
-
339
- folders = folder_names_and_paths[folder_name]
340
- for x in folders[0]:
341
- if os.path.isdir(x):
342
- if x not in out[1]:
343
- return None
344
-
345
- return out
346
-
347
- def get_filename_list(folder_name: str) -> list[str]:
348
- folder_name = map_legacy(folder_name)
349
- out = cached_filename_list_(folder_name)
350
- if out is None:
351
- out = get_filename_list_(folder_name)
352
- global filename_list_cache
353
- filename_list_cache[folder_name] = out
354
- cache_helper.set(folder_name, out)
355
- return list(out[0])
356
-
357
- def get_save_image_path(filename_prefix: str, output_dir: str, image_width=0, image_height=0) -> tuple[str, str, int, str, str]:
358
- def map_filename(filename: str) -> tuple[int, str]:
359
- prefix_len = len(os.path.basename(filename_prefix))
360
- prefix = filename[:prefix_len + 1]
361
- try:
362
- digits = int(filename[prefix_len + 1:].split('_')[0])
363
- except:
364
- digits = 0
365
- return digits, prefix
366
-
367
- def compute_vars(input: str, image_width: int, image_height: int) -> str:
368
- input = input.replace("%width%", str(image_width))
369
- input = input.replace("%height%", str(image_height))
370
- now = time.localtime()
371
- input = input.replace("%year%", str(now.tm_year))
372
- input = input.replace("%month%", str(now.tm_mon).zfill(2))
373
- input = input.replace("%day%", str(now.tm_mday).zfill(2))
374
- input = input.replace("%hour%", str(now.tm_hour).zfill(2))
375
- input = input.replace("%minute%", str(now.tm_min).zfill(2))
376
- input = input.replace("%second%", str(now.tm_sec).zfill(2))
377
- return input
378
-
379
- if "%" in filename_prefix:
380
- filename_prefix = compute_vars(filename_prefix, image_width, image_height)
381
-
382
- subfolder = os.path.dirname(os.path.normpath(filename_prefix))
383
- filename = os.path.basename(os.path.normpath(filename_prefix))
384
-
385
- full_output_folder = os.path.join(output_dir, subfolder)
386
-
387
- if os.path.commonpath((output_dir, os.path.abspath(full_output_folder))) != output_dir:
388
- err = "**** ERROR: Saving image outside the output folder is not allowed." + \
389
- "\n full_output_folder: " + os.path.abspath(full_output_folder) + \
390
- "\n output_dir: " + output_dir + \
391
- "\n commonpath: " + os.path.commonpath((output_dir, os.path.abspath(full_output_folder)))
392
- logging.error(err)
393
- raise Exception(err)
394
-
395
- try:
396
- counter = max(filter(lambda a: os.path.normcase(a[1][:-1]) == os.path.normcase(filename) and a[1][-1] == "_", map(map_filename, os.listdir(full_output_folder))))[0] + 1
397
- except ValueError:
398
- counter = 1
399
- except FileNotFoundError:
400
- os.makedirs(full_output_folder, exist_ok=True)
401
- counter = 1
402
- return full_output_folder, filename, counter, subfolder, filename_prefix
403
-
404
- def get_input_subfolders() -> list[str]:
405
- """Returns a list of all subfolder paths in the input directory, recursively.
406
-
407
- Returns:
408
- List of folder paths relative to the input directory, excluding the root directory
409
- """
410
- input_dir = get_input_directory()
411
- folders = []
412
-
413
- try:
414
- if not os.path.exists(input_dir):
415
- return []
416
-
417
- for root, dirs, _ in os.walk(input_dir):
418
- rel_path = os.path.relpath(root, input_dir)
419
- if rel_path != ".": # Only include non-root directories
420
- # Normalize path separators to forward slashes
421
- folders.append(rel_path.replace(os.sep, '/'))
422
-
423
- return sorted(folders)
424
- except FileNotFoundError:
425
- return []