Integration of from_pretrained and from_single_file

src/diffusers/loaders/single_file.py

@@ -314,6 +314,8 @@ def from_single_file(cls, pretrained_model_link_or_path, **kwargs) -> Self:
             disable_mmap ('bool', *optional*, defaults to 'False'):
                 Whether to disable mmap when loading a Safetensors model. This option can perform better when the model
                 is on a network mount or hard drive.
+            checkpoint (`dict`, *optional*):
+                The loaded state dictionary of the model.
             kwargs (remaining dictionary of keyword arguments, *optional*):
                 Can be used to overwrite load and saveable variables (the pipeline components of the specific pipeline
                 class). The overwritten components are passed directly to the pipelines `__init__` method. See example
@@ -362,6 +364,7 @@ def from_single_file(cls, pretrained_model_link_or_path, **kwargs) -> Self:
         revision = kwargs.pop("revision", None)
         torch_dtype = kwargs.pop("torch_dtype", torch.float32)
         disable_mmap = kwargs.pop("disable_mmap", False)
+        checkpoint = kwargs.pop("checkpoint", None)
 
         is_legacy_loading = False
 
@@ -386,18 +389,19 @@ def from_single_file(cls, pretrained_model_link_or_path, **kwargs) -> Self:
 
         from ..pipelines.pipeline_utils import _get_pipeline_class
 
-        pipeline_class = _get_pipeline_class(cls, config=None)
-
-        checkpoint = load_single_file_checkpoint(
-            pretrained_model_link_or_path,
-            force_download=force_download,
-            proxies=proxies,
-            token=token,
-            cache_dir=cache_dir,
-            local_files_only=local_files_only,
-            revision=revision,
-            disable_mmap=disable_mmap,
-        )
+        pipeline_class = _get_pipeline_class(cls, class_name=cls.__name__, config=None)
+
+        if checkpoint is None:
+            checkpoint = load_single_file_checkpoint(
+                pretrained_model_link_or_path,
+                force_download=force_download,
+                proxies=proxies,
+                token=token,
+                cache_dir=cache_dir,
+                local_files_only=local_files_only,
+                revision=revision,
+                disable_mmap=disable_mmap,
+            )
 
         if config is None:
             config = fetch_diffusers_config(checkpoint)
@@ -480,6 +484,11 @@ def from_single_file(cls, pretrained_model_link_or_path, **kwargs) -> Self:
         init_kwargs = {k: init_dict.pop(k) for k in optional_kwargs if k in init_dict}
         init_kwargs = {**init_kwargs, **passed_pipe_kwargs}
 
+        if len(unused_kwargs) > 0:
+            logger.warning(
+                f"Keyword arguments {unused_kwargs} are not expected by {pipeline_class.__name__} and will be ignored."
+            )
+
         from diffusers import pipelines
 
         # remove `null` components

src/diffusers/loaders/single_file_utils.py

@@ -2857,3 +2857,83 @@ def convert_lumina_attn_to_diffusers(tensor, diffusers_key):
             converted_state_dict[diffusers_key] = checkpoint.pop(key)
 
     return converted_state_dict
+
+
+def get_keyword_types(keyword):
+    r"""
+    Determine the type and loading method for a given keyword.
+
+    Parameters:
+        keyword (`str`):
+            The input keyword to classify.
+
+    Returns:
+        `dict`: A dictionary containing the model format, loading method,
+                and various types and extra types flags.
+    """
+
+    # Initialize the status dictionary with default values
+    status = {
+        "checkpoint_format": None,
+        "loading_method": None,
+        "type": {
+            "other": False,
+            "hf_url": False,
+            "hf_repo": False,
+            "civitai_url": False,
+            "local": False,
+        },
+        "extra_type": {
+            "url": False,
+            "missing_model_index": None,
+        },
+    }
+
+    # Check if the keyword is an HTTP or HTTPS URL
+    status["extra_type"]["url"] = bool(re.search(r"^(https?)://", keyword))
+
+    # Check if the keyword is a file
+    if os.path.isfile(keyword):
+        status["type"]["local"] = True
+        status["checkpoint_format"] = "single_file"
+        status["loading_method"] = "from_single_file"
+
+    # Check if the keyword is a directory
+    elif os.path.isdir(keyword):
+        status["type"]["local"] = True
+        status["checkpoint_format"] = "diffusers"
+        status["loading_method"] = "from_pretrained"
+        if not os.path.exists(os.path.join(keyword, "model_index.json")):
+            status["extra_type"]["missing_model_index"] = True
+
+    # Check if the keyword is a Civitai URL
+    elif keyword.startswith("https://civitai.com/"):
+        status["type"]["civitai_url"] = True
+        status["checkpoint_format"] = "single_file"
+        status["loading_method"] = None
+
+    # Check if the keyword starts with any valid URL prefixes
+    elif any(keyword.startswith(prefix) for prefix in VALID_URL_PREFIXES):
+        repo_id, weights_name = _extract_repo_id_and_weights_name(keyword)
+        if weights_name:
+            status["type"]["hf_url"] = True
+            status["checkpoint_format"] = "single_file"
+            status["loading_method"] = "from_single_file"
+        else:
+            status["type"]["hf_repo"] = True
+            status["checkpoint_format"] = "diffusers"
+            status["loading_method"] = "from_pretrained"
+
+    # Check if the keyword matches a Hugging Face repository format
+    elif re.match(r"^[^/]+/[^/]+$", keyword):
+        status["type"]["hf_repo"] = True
+        status["checkpoint_format"] = "diffusers"
+        status["loading_method"] = "from_pretrained"
+
+    # If none of the above apply
+    else:
+        status["type"]["other"] = True
+        status["checkpoint_format"] = None
+        status["loading_method"] = None
+
+    return status