Model manager route enhancements

invokeai/app/api/routers/models.py

@@ -1,6 +1,7 @@
-# Copyright (c) 2023 Kyle Schouviller (https://github.com/kyle0654), 2023 Kent Keirsey (https://github.com/hipsterusername), 2024 Lincoln Stein
+# Copyright (c) 2023 Kyle Schouviller (https://github.com/kyle0654), 2023 Kent Keirsey (https://github.com/hipsterusername), 2023 Lincoln D. Stein
 
 
+import pathlib
 from typing import Literal, List, Optional, Union
 
 from fastapi import Body, Path, Query, Response
@@ -22,6 +23,7 @@
 ImportModelResponse = Union[tuple(OPENAPI_MODEL_CONFIGS)]
 ConvertModelResponse = Union[tuple(OPENAPI_MODEL_CONFIGS)]
 MergeModelResponse = Union[tuple(OPENAPI_MODEL_CONFIGS)]
+ImportModelAttributes = Union[tuple(OPENAPI_MODEL_CONFIGS)]
 
 class ModelsList(BaseModel):
     models: list[Union[tuple(OPENAPI_MODEL_CONFIGS)]]
@@ -78,7 +80,7 @@ async def update_model(
     return model_response
 
 @models_router.post(
-    "/",
+    "/import",
     operation_id="import_model",
     responses= {
         201: {"description" : "The model imported successfully"},
@@ -94,7 +96,7 @@ async def import_model(
         prediction_type: Optional[Literal['v_prediction','epsilon','sample']] = \
                 Body(description='Prediction type for SDv2 checkpoint files', default="v_prediction"),
 ) -> ImportModelResponse:
-    """ Add a model using its local path, repo_id, or remote URL """
+    """ Add a model using its local path, repo_id, or remote URL. Model characteristics will be probed and configured automatically """
 
     items_to_import = {location}
     prediction_types = { x.value: x for x in SchedulerPredictionType }
@@ -126,18 +128,100 @@ async def import_model(
         logger.error(str(e))
         raise HTTPException(status_code=409, detail=str(e))
 
+@models_router.post(
+    "/add",
+    operation_id="add_model",
+    responses= {
+        201: {"description" : "The model added successfully"},
+        404: {"description" : "The model could not be found"},
+        424: {"description" : "The model appeared to add successfully, but could not be found in the model manager"},
+        409: {"description" : "There is already a model corresponding to this path or repo_id"},
+    },
+    status_code=201,
+    response_model=ImportModelResponse
+)
+async def add_model(
+        info: Union[tuple(OPENAPI_MODEL_CONFIGS)] = Body(description="Model configuration"),
+) -> ImportModelResponse:
+    """ Add a model using the configuration information appropriate for its type. Only local models can be added by path"""
+
+    logger = ApiDependencies.invoker.services.logger
+
+    try:
+        ApiDependencies.invoker.services.model_manager.add_model(
+            info.model_name,
+            info.base_model,
+            info.model_type,
+            model_attributes = info.dict()
+        )
+        logger.info(f'Successfully added {info.model_name}')
+        model_raw = ApiDependencies.invoker.services.model_manager.list_model(
+            model_name=info.model_name,
+            base_model=info.base_model,
+            model_type=info.model_type
+        )
+        return parse_obj_as(ImportModelResponse, model_raw)
+    except KeyError as e:
+        logger.error(str(e))
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        logger.error(str(e))
+        raise HTTPException(status_code=409, detail=str(e))
 
+@models_router.post(
+    "/rename/{base_model}/{model_type}/{model_name}",
+    operation_id="rename_model",
+    responses= {
+        201: {"description" : "The model was renamed successfully"},
+        404: {"description" : "The model could not be found"},
+        409: {"description" : "There is already a model corresponding to the new name"},
+    },
+    status_code=201,
+    response_model=ImportModelResponse
+)
+async def rename_model(
+        base_model: BaseModelType = Path(description="Base model"),
+        model_type: ModelType = Path(description="The type of model"),
+        model_name: str = Path(description="current model name"),
+        new_name: Optional[str] = Query(description="new model name", default=None),
+        new_base: Optional[BaseModelType] = Query(description="new model base", default=None),
+) -> ImportModelResponse:
+    """ Rename a model"""
+
+    logger = ApiDependencies.invoker.services.logger
+
+    try:
+        result = ApiDependencies.invoker.services.model_manager.rename_model(
+            base_model = base_model,
+            model_type = model_type,
+            model_name = model_name,
+            new_name = new_name,
+            new_base = new_base,
+        )
+        logger.debug(result)
+        logger.info(f'Successfully renamed {model_name}=>{new_name}')
+        model_raw = ApiDependencies.invoker.services.model_manager.list_model(
+            model_name=new_name or model_name,
+            base_model=new_base or base_model,
+            model_type=model_type
+        )
+        return parse_obj_as(ImportModelResponse, model_raw)
+    except KeyError as e:
+        logger.error(str(e))
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        logger.error(str(e))
+        raise HTTPException(status_code=409, detail=str(e))
+
 @models_router.delete(
     "/{base_model}/{model_type}/{model_name}",
     operation_id="del_model",
     responses={
-        204: {
-        "description": "Model deleted successfully"
-        }, 
-        404: {
-        "description": "Model not found"
-        }
+        204: { "description": "Model deleted successfully" }, 
+        404: { "description": "Model not found" }
     },
+    status_code = 204,
+    response_model = None,
 )
 async def delete_model(
         base_model: BaseModelType = Path(description="Base model"),
@@ -173,14 +257,17 @@ async def convert_model(
         base_model: BaseModelType = Path(description="Base model"),
         model_type: ModelType = Path(description="The type of model"),
         model_name: str = Path(description="model name"),
+        convert_dest_directory: Optional[str] = Query(default=None, description="Save the converted model to the designated directory"),
 ) -> ConvertModelResponse:
-    """Convert a checkpoint model into a diffusers model"""
+    """Convert a checkpoint model into a diffusers model, optionally saving to the indicated destination directory, or `models` if none."""
     logger = ApiDependencies.invoker.services.logger
     try:
         logger.info(f"Converting model: {model_name}")
+        dest = pathlib.Path(convert_dest_directory) if convert_dest_directory else None
         ApiDependencies.invoker.services.model_manager.convert_model(model_name,
                                                                      base_model = base_model,
-                                                                     model_type = model_type
+                                                                     model_type = model_type,
+                                                                     convert_dest_directory = dest,
                                                                      )
         model_raw = ApiDependencies.invoker.services.model_manager.list_model(model_name,
                                                                               base_model = base_model,
@@ -191,6 +278,53 @@ async def convert_model(
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
     return response
+
+@models_router.get(
+    "/search",
+    operation_id="search_for_models",
+    responses={
+        200: { "description": "Directory searched successfully" },
+        404: { "description": "Invalid directory path"  },
+    },
+    status_code = 200,
+    response_model = List[pathlib.Path]
+)
+async def search_for_models(
+        search_path: pathlib.Path = Query(description="Directory path to search for models")
+)->List[pathlib.Path]:
+    if not search_path.is_dir():
+        raise HTTPException(status_code=404, detail=f"The search path '{search_path}' does not exist or is not directory")
+    return ApiDependencies.invoker.services.model_manager.search_for_models([search_path])
+
+@models_router.get(
+    "/ckpt_confs",
+    operation_id="list_ckpt_configs",
+    responses={
+        200: { "description" : "paths retrieved successfully" },
+    },
+    status_code = 200,
+    response_model = List[pathlib.Path]
+)
+async def list_ckpt_configs(
+)->List[pathlib.Path]:
+    """Return a list of the legacy checkpoint configuration files stored in `ROOT/configs/stable-diffusion`, relative to ROOT."""
+    return ApiDependencies.invoker.services.model_manager.list_checkpoint_configs()
+
+
+@models_router.get(
+    "/sync",
+    operation_id="sync_to_config",
+    responses={
+        201: { "description": "synchronization successful" },
+    },
+    status_code = 201,
+    response_model = None
+)
+async def sync_to_config(
+)->None:
+    """Call after making changes to models.yaml, autoimport directories or models directory to synchronize
+    in-memory data structures with disk data structures."""
+    return ApiDependencies.invoker.services.model_manager.sync_to_config()
 
 @models_router.put(
     "/merge/{base_model}",
@@ -210,17 +344,21 @@ async def merge_models(
         alpha: Optional[float]                     = Body(description="Alpha weighting strength to apply to 2d and 3d models", default=0.5),
         interp: Optional[MergeInterpolationMethod] = Body(description="Interpolation method"),
         force: Optional[bool]                      = Body(description="Force merging of models created with different versions of diffusers", default=False),
+        merge_dest_directory: Optional[str]       = Body(description="Save the merged model to the designated directory (with 'merged_model_name' appended)", default=None)
 ) -> MergeModelResponse:
     """Convert a checkpoint model into a diffusers model"""
     logger = ApiDependencies.invoker.services.logger
     try:
-        logger.info(f"Merging models: {model_names}")
+        logger.info(f"Merging models: {model_names} into {merge_dest_directory or '<MODELS>'}/{merged_model_name}")
+        dest = pathlib.Path(merge_dest_directory) if merge_dest_directory else None
         result = ApiDependencies.invoker.services.model_manager.merge_models(model_names,
                                                                              base_model,
-                                                                             merged_model_name or "+".join(model_names),
-                                                                             alpha,
-                                                                             interp,
-                                                                             force)
+                                                                             merged_model_name=merged_model_name or "+".join(model_names),
+                                                                             alpha=alpha,
+                                                                             interp=interp,
+                                                                             force=force,
+                                                                             merge_dest_directory = dest
+                                                                             )
         model_raw = ApiDependencies.invoker.services.model_manager.list_model(result.name,
                                                                               base_model = base_model,
                                                                               model_type = ModelType.Main,