We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
BUG: [UI] image model does not show `cache` status
Note that the issue tracker is NOT the place for general support.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `xinference/model/image/core.py`
Content:
```
1 # Copyright 2022-2023 XProbe Inc.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 import collections.abc
15 import logging
16 import os
17 from collections import defaultdict
18 from typing import Dict, List, Optional, Tuple
19 
20 from ...constants import XINFERENCE_CACHE_DIR
21 from ..core import CacheableModelSpec, ModelDescription
22 from ..utils import valid_model_revision
23 from .stable_diffusion.core import DiffusionModel
24 
25 MAX_ATTEMPTS = 3
26 
27 logger = logging.getLogger(__name__)
28 
29 IMAGE_MODEL_DESCRIPTIONS: Dict[str, List[Dict]] = defaultdict(list)
30 
31 
32 def get_image_model_descriptions():
33     import copy
34 
35     return copy.deepcopy(IMAGE_MODEL_DESCRIPTIONS)
36 
37 
38 class ImageModelFamilyV1(CacheableModelSpec):
39     model_family: str
40     model_name: str
41     model_id: str
42     model_revision: str
43     model_hub: str = "huggingface"
44     controlnet: Optional[List["ImageModelFamilyV1"]]
45 
46 
47 class ImageModelDescription(ModelDescription):
48     def __init__(
49         self,
50         address: Optional[str],
51         devices: Optional[List[str]],
52         model_spec: ImageModelFamilyV1,
53         model_path: Optional[str] = None,
54     ):
55         super().__init__(address, devices, model_path=model_path)
56         self._model_spec = model_spec
57 
58     def to_dict(self):
59         if self._model_spec.controlnet is not None:
60             controlnet = [cn.dict() for cn in self._model_spec.controlnet]
61         else:
62             controlnet = self._model_spec.controlnet
63         return {
64             "model_type": "image",
65             "address": self.address,
66             "accelerators": self.devices,
67             "model_name": self._model_spec.model_name,
68             "model_family": self._model_spec.model_family,
69             "model_revision": self._model_spec.model_revision,
70             "controlnet": controlnet,
71         }
72 
73     def to_version_info(self):
74         from .utils import get_model_version
75 
76         if self._model_path is None:
77             is_cached = get_cache_status(self._model_spec)
78             file_location = get_cache_dir(self._model_spec)
79         else:
80             is_cached = True
81             file_location = self._model_path
82 
83         if self._model_spec.controlnet is None:
84             return [
85                 {
86                     "model_version": get_model_version(self._model_spec, None),
87                     "model_file_location": file_location,
88                     "cache_status": is_cached,
89                     "controlnet": "zoe-depth",
90                 }
91             ]
92         else:
93             res = []
94             for cn in self._model_spec.controlnet:
95                 res.append(
96                     {
97                         "model_version": get_model_version(self._model_spec, cn),
98                         "model_file_location": file_location,
99                         "cache_status": is_cached,
100                         "controlnet": cn.model_name,
101                     }
102                 )
103             return res
104 
105 
106 def generate_image_description(
107     image_model: ImageModelFamilyV1,
108 ) -> Dict[str, List[Dict]]:
109     res = defaultdict(list)
110     res[image_model.model_name].extend(
111         ImageModelDescription(None, None, image_model).to_version_info()
112     )
113     return res
114 
115 
116 def match_diffusion(model_name: str) -> ImageModelFamilyV1:
117     from ..utils import download_from_modelscope
118     from . import BUILTIN_IMAGE_MODELS, MODELSCOPE_IMAGE_MODELS
119 
120     if download_from_modelscope():
121         if model_name in MODELSCOPE_IMAGE_MODELS:
122             logger.debug(f"Image model {model_name} found in ModelScope.")
123             return MODELSCOPE_IMAGE_MODELS[model_name]
124         else:
125             logger.debug(
126                 f"Image model {model_name} not found in ModelScope, "
127                 f"now try to load it via builtin way."
128             )
129 
130     if model_name in BUILTIN_IMAGE_MODELS:
131         return BUILTIN_IMAGE_MODELS[model_name]
132     else:
133         raise ValueError(
134             f"Image model {model_name} not found, available"
135             f"model list: {BUILTIN_IMAGE_MODELS.keys()}"
136         )
137 
138 
139 def cache(model_spec: ImageModelFamilyV1):
140     from ..utils import cache
141 
142     return cache(model_spec, ImageModelDescription)
143 
144 
145 def get_cache_dir(model_spec: ImageModelFamilyV1):
146     return os.path.realpath(os.path.join(XINFERENCE_CACHE_DIR, model_spec.model_name))
147 
148 
149 def get_cache_status(
150     model_spec: ImageModelFamilyV1,
151 ) -> bool:
152     cache_dir = get_cache_dir(model_spec)
153     meta_path = os.path.join(cache_dir, "__valid_download")
154     return valid_model_revision(meta_path, model_spec.model_revision)
155 
156 
157 def create_image_model_instance(
158     subpool_addr: str,
159     devices: List[str],
160     model_uid: str,
161     model_name: str,
162     lora_model_path: Optional[str] = None,
163     lora_load_kwargs: Optional[Dict] = None,
164     lora_fuse_kwargs: Optional[Dict] = None,
165     **kwargs,
166 ) -> Tuple[DiffusionModel, ImageModelDescription]:
167     model_spec = match_diffusion(model_name)
168     controlnet = kwargs.get("controlnet")
169     # Handle controlnet
170     if controlnet is not None:
171         if isinstance(controlnet, str):
172             controlnet = [controlnet]
173         elif not isinstance(controlnet, collections.abc.Sequence):
174             raise ValueError("controlnet should be a str or a list of str.")
175         elif set(controlnet) != len(controlnet):
176             raise ValueError("controlnet should be a list of unique str.")
177         elif not model_spec.controlnet:
178             raise ValueError(f"Model {model_name} has empty controlnet list.")
179 
180         controlnet_model_paths = []
181         assert model_spec.controlnet is not None
182         for name in controlnet:
183             for cn_model_spec in model_spec.controlnet:
184                 if cn_model_spec.model_name == name:
185                     model_path = cache(cn_model_spec)
186                     controlnet_model_paths.append(model_path)
187                     break
188             else:
189                 raise ValueError(
190                     f"controlnet `{name}` is not supported for model `{model_name}`."
191                 )
192         if len(controlnet_model_paths) == 1:
193             kwargs["controlnet"] = controlnet_model_paths[0]
194         else:
195             kwargs["controlnet"] = controlnet_model_paths
196     model_path = cache(model_spec)
197     model = DiffusionModel(
198         model_uid,
199         model_path,
200         lora_model_path=lora_model_path,
201         lora_load_kwargs=lora_load_kwargs,
202         lora_fuse_kwargs=lora_fuse_kwargs,
203         **kwargs,
204     )
205     model_description = ImageModelDescription(
206         subpool_addr, devices, model_spec, model_path=model_path
207     )
208     return model, model_description
209
```

Path: `xinference/model/image/__init__.py`
Content:
```
1 # Copyright 2022-2023 XProbe Inc.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import codecs
16 import json
17 import os
18 from itertools import chain
19 
20 from .core import (
21     IMAGE_MODEL_DESCRIPTIONS,
22     ImageModelFamilyV1,
23     generate_image_description,
24     get_cache_status,
25     get_image_model_descriptions,
26 )
27 
28 _model_spec_json = os.path.join(os.path.dirname(__file__), "model_spec.json")
29 _model_spec_modelscope_json = os.path.join(
30     os.path.dirname(__file__), "model_spec_modelscope.json"
31 )
32 BUILTIN_IMAGE_MODELS = dict(
33     (spec["model_name"], ImageModelFamilyV1(**spec))
34     for spec in json.load(codecs.open(_model_spec_json, "r", encoding="utf-8"))
35 )
36 MODELSCOPE_IMAGE_MODELS = dict(
37     (spec["model_name"], ImageModelFamilyV1(**spec))
38     for spec in json.load(
39         codecs.open(_model_spec_modelscope_json, "r", encoding="utf-8")
40     )
41 )
42 
43 # register model description
44 for model_name, model_spec in chain(
45     MODELSCOPE_IMAGE_MODELS.items(), BUILTIN_IMAGE_MODELS.items()
46 ):
47     IMAGE_MODEL_DESCRIPTIONS.update(generate_image_description(model_spec))
48 
49 del _model_spec_json
50
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/xinference/model/image/__init__.py b/xinference/model/image/__init__.py
--- a/xinference/model/image/__init__.py
+++ b/xinference/model/image/__init__.py
@@ -18,7 +18,9 @@
 from itertools import chain
 
 from .core import (
+    BUILTIN_IMAGE_MODELS,
     IMAGE_MODEL_DESCRIPTIONS,
+    MODELSCOPE_IMAGE_MODELS,
     ImageModelFamilyV1,
     generate_image_description,
     get_cache_status,
@@ -29,14 +31,18 @@
 _model_spec_modelscope_json = os.path.join(
     os.path.dirname(__file__), "model_spec_modelscope.json"
 )
-BUILTIN_IMAGE_MODELS = dict(
-    (spec["model_name"], ImageModelFamilyV1(**spec))
-    for spec in json.load(codecs.open(_model_spec_json, "r", encoding="utf-8"))
+BUILTIN_IMAGE_MODELS.update(
+    dict(
+        (spec["model_name"], ImageModelFamilyV1(**spec))
+        for spec in json.load(codecs.open(_model_spec_json, "r", encoding="utf-8"))
+    )
 )
-MODELSCOPE_IMAGE_MODELS = dict(
-    (spec["model_name"], ImageModelFamilyV1(**spec))
-    for spec in json.load(
-        codecs.open(_model_spec_modelscope_json, "r", encoding="utf-8")
+MODELSCOPE_IMAGE_MODELS.update(
+    dict(
+        (spec["model_name"], ImageModelFamilyV1(**spec))
+        for spec in json.load(
+            codecs.open(_model_spec_modelscope_json, "r", encoding="utf-8")
+        )
     )
 )
 
diff --git a/xinference/model/image/core.py b/xinference/model/image/core.py
--- a/xinference/model/image/core.py
+++ b/xinference/model/image/core.py
@@ -27,6 +27,8 @@
 logger = logging.getLogger(__name__)
 
 IMAGE_MODEL_DESCRIPTIONS: Dict[str, List[Dict]] = defaultdict(list)
+BUILTIN_IMAGE_MODELS: Dict[str, "ImageModelFamilyV1"] = {}
+MODELSCOPE_IMAGE_MODELS: Dict[str, "ImageModelFamilyV1"] = {}
 
 
 def get_image_model_descriptions():
@@ -151,7 +153,21 @@
 ) -> bool:
     cache_dir = get_cache_dir(model_spec)
     meta_path = os.path.join(cache_dir, "__valid_download")
-    return valid_model_revision(meta_path, model_spec.model_revision)
+
+    model_name = model_spec.model_name
+    if model_name in BUILTIN_IMAGE_MODELS and model_name in MODELSCOPE_IMAGE_MODELS:
+        hf_spec = BUILTIN_IMAGE_MODELS[model_name]
+        ms_spec = MODELSCOPE_IMAGE_MODELS[model_name]
+
+        return any(
+            [
+                valid_model_revision(meta_path, hf_spec.model_revision),
+                valid_model_revision(meta_path, ms_spec.model_revision),
+            ]
+        )
+    else:  # Usually for UT
+        logger.warning(f"Cannot find builtin image model spec: {model_name}")
+        return valid_model_revision(meta_path, model_spec.model_revision)
 
 
 def create_image_model_instance(

{"golden_diff": "diff --git a/xinference/model/image/__init__.py b/xinference/model/image/__init__.py\n--- a/xinference/model/image/__init__.py\n+++ b/xinference/model/image/__init__.py\n@@ -18,7 +18,9 @@\n from itertools import chain\n \n from .core import (\n+    BUILTIN_IMAGE_MODELS,\n     IMAGE_MODEL_DESCRIPTIONS,\n+    MODELSCOPE_IMAGE_MODELS,\n     ImageModelFamilyV1,\n     generate_image_description,\n     get_cache_status,\n@@ -29,14 +31,18 @@\n _model_spec_modelscope_json = os.path.join(\n     os.path.dirname(__file__), \"model_spec_modelscope.json\"\n )\n-BUILTIN_IMAGE_MODELS = dict(\n-    (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n-    for spec in json.load(codecs.open(_model_spec_json, \"r\", encoding=\"utf-8\"))\n+BUILTIN_IMAGE_MODELS.update(\n+    dict(\n+        (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n+        for spec in json.load(codecs.open(_model_spec_json, \"r\", encoding=\"utf-8\"))\n+    )\n )\n-MODELSCOPE_IMAGE_MODELS = dict(\n-    (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n-    for spec in json.load(\n-        codecs.open(_model_spec_modelscope_json, \"r\", encoding=\"utf-8\")\n+MODELSCOPE_IMAGE_MODELS.update(\n+    dict(\n+        (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n+        for spec in json.load(\n+            codecs.open(_model_spec_modelscope_json, \"r\", encoding=\"utf-8\")\n+        )\n     )\n )\n \ndiff --git a/xinference/model/image/core.py b/xinference/model/image/core.py\n--- a/xinference/model/image/core.py\n+++ b/xinference/model/image/core.py\n@@ -27,6 +27,8 @@\n logger = logging.getLogger(__name__)\n \n IMAGE_MODEL_DESCRIPTIONS: Dict[str, List[Dict]] = defaultdict(list)\n+BUILTIN_IMAGE_MODELS: Dict[str, \"ImageModelFamilyV1\"] = {}\n+MODELSCOPE_IMAGE_MODELS: Dict[str, \"ImageModelFamilyV1\"] = {}\n \n \n def get_image_model_descriptions():\n@@ -151,7 +153,21 @@\n ) -> bool:\n     cache_dir = get_cache_dir(model_spec)\n     meta_path = os.path.join(cache_dir, \"__valid_download\")\n-    return valid_model_revision(meta_path, model_spec.model_revision)\n+\n+    model_name = model_spec.model_name\n+    if model_name in BUILTIN_IMAGE_MODELS and model_name in MODELSCOPE_IMAGE_MODELS:\n+        hf_spec = BUILTIN_IMAGE_MODELS[model_name]\n+        ms_spec = MODELSCOPE_IMAGE_MODELS[model_name]\n+\n+        return any(\n+            [\n+                valid_model_revision(meta_path, hf_spec.model_revision),\n+                valid_model_revision(meta_path, ms_spec.model_revision),\n+            ]\n+        )\n+    else:  # Usually for UT\n+        logger.warning(f\"Cannot find builtin image model spec: {model_name}\")\n+        return valid_model_revision(meta_path, model_spec.model_revision)\n \n \n def create_image_model_instance(\n", "issue": "BUG: [UI] image model does not show `cache` status\nNote that the issue tracker is NOT the place for general support.\r\n\n", "before_files": [{"content": "# Copyright 2022-2023 XProbe Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport collections.abc\nimport logging\nimport os\nfrom collections import defaultdict\nfrom typing import Dict, List, Optional, Tuple\n\nfrom ...constants import XINFERENCE_CACHE_DIR\nfrom ..core import CacheableModelSpec, ModelDescription\nfrom ..utils import valid_model_revision\nfrom .stable_diffusion.core import DiffusionModel\n\nMAX_ATTEMPTS = 3\n\nlogger = logging.getLogger(__name__)\n\nIMAGE_MODEL_DESCRIPTIONS: Dict[str, List[Dict]] = defaultdict(list)\n\n\ndef get_image_model_descriptions():\n    import copy\n\n    return copy.deepcopy(IMAGE_MODEL_DESCRIPTIONS)\n\n\nclass ImageModelFamilyV1(CacheableModelSpec):\n    model_family: str\n    model_name: str\n    model_id: str\n    model_revision: str\n    model_hub: str = \"huggingface\"\n    controlnet: Optional[List[\"ImageModelFamilyV1\"]]\n\n\nclass ImageModelDescription(ModelDescription):\n    def __init__(\n        self,\n        address: Optional[str],\n        devices: Optional[List[str]],\n        model_spec: ImageModelFamilyV1,\n        model_path: Optional[str] = None,\n    ):\n        super().__init__(address, devices, model_path=model_path)\n        self._model_spec = model_spec\n\n    def to_dict(self):\n        if self._model_spec.controlnet is not None:\n            controlnet = [cn.dict() for cn in self._model_spec.controlnet]\n        else:\n            controlnet = self._model_spec.controlnet\n        return {\n            \"model_type\": \"image\",\n            \"address\": self.address,\n            \"accelerators\": self.devices,\n            \"model_name\": self._model_spec.model_name,\n            \"model_family\": self._model_spec.model_family,\n            \"model_revision\": self._model_spec.model_revision,\n            \"controlnet\": controlnet,\n        }\n\n    def to_version_info(self):\n        from .utils import get_model_version\n\n        if self._model_path is None:\n            is_cached = get_cache_status(self._model_spec)\n            file_location = get_cache_dir(self._model_spec)\n        else:\n            is_cached = True\n            file_location = self._model_path\n\n        if self._model_spec.controlnet is None:\n            return [\n                {\n                    \"model_version\": get_model_version(self._model_spec, None),\n                    \"model_file_location\": file_location,\n                    \"cache_status\": is_cached,\n                    \"controlnet\": \"zoe-depth\",\n                }\n            ]\n        else:\n            res = []\n            for cn in self._model_spec.controlnet:\n                res.append(\n                    {\n                        \"model_version\": get_model_version(self._model_spec, cn),\n                        \"model_file_location\": file_location,\n                        \"cache_status\": is_cached,\n                        \"controlnet\": cn.model_name,\n                    }\n                )\n            return res\n\n\ndef generate_image_description(\n    image_model: ImageModelFamilyV1,\n) -> Dict[str, List[Dict]]:\n    res = defaultdict(list)\n    res[image_model.model_name].extend(\n        ImageModelDescription(None, None, image_model).to_version_info()\n    )\n    return res\n\n\ndef match_diffusion(model_name: str) -> ImageModelFamilyV1:\n    from ..utils import download_from_modelscope\n    from . import BUILTIN_IMAGE_MODELS, MODELSCOPE_IMAGE_MODELS\n\n    if download_from_modelscope():\n        if model_name in MODELSCOPE_IMAGE_MODELS:\n            logger.debug(f\"Image model {model_name} found in ModelScope.\")\n            return MODELSCOPE_IMAGE_MODELS[model_name]\n        else:\n            logger.debug(\n                f\"Image model {model_name} not found in ModelScope, \"\n                f\"now try to load it via builtin way.\"\n            )\n\n    if model_name in BUILTIN_IMAGE_MODELS:\n        return BUILTIN_IMAGE_MODELS[model_name]\n    else:\n        raise ValueError(\n            f\"Image model {model_name} not found, available\"\n            f\"model list: {BUILTIN_IMAGE_MODELS.keys()}\"\n        )\n\n\ndef cache(model_spec: ImageModelFamilyV1):\n    from ..utils import cache\n\n    return cache(model_spec, ImageModelDescription)\n\n\ndef get_cache_dir(model_spec: ImageModelFamilyV1):\n    return os.path.realpath(os.path.join(XINFERENCE_CACHE_DIR, model_spec.model_name))\n\n\ndef get_cache_status(\n    model_spec: ImageModelFamilyV1,\n) -> bool:\n    cache_dir = get_cache_dir(model_spec)\n    meta_path = os.path.join(cache_dir, \"__valid_download\")\n    return valid_model_revision(meta_path, model_spec.model_revision)\n\n\ndef create_image_model_instance(\n    subpool_addr: str,\n    devices: List[str],\n    model_uid: str,\n    model_name: str,\n    lora_model_path: Optional[str] = None,\n    lora_load_kwargs: Optional[Dict] = None,\n    lora_fuse_kwargs: Optional[Dict] = None,\n    **kwargs,\n) -> Tuple[DiffusionModel, ImageModelDescription]:\n    model_spec = match_diffusion(model_name)\n    controlnet = kwargs.get(\"controlnet\")\n    # Handle controlnet\n    if controlnet is not None:\n        if isinstance(controlnet, str):\n            controlnet = [controlnet]\n        elif not isinstance(controlnet, collections.abc.Sequence):\n            raise ValueError(\"controlnet should be a str or a list of str.\")\n        elif set(controlnet) != len(controlnet):\n            raise ValueError(\"controlnet should be a list of unique str.\")\n        elif not model_spec.controlnet:\n            raise ValueError(f\"Model {model_name} has empty controlnet list.\")\n\n        controlnet_model_paths = []\n        assert model_spec.controlnet is not None\n        for name in controlnet:\n            for cn_model_spec in model_spec.controlnet:\n                if cn_model_spec.model_name == name:\n                    model_path = cache(cn_model_spec)\n                    controlnet_model_paths.append(model_path)\n                    break\n            else:\n                raise ValueError(\n                    f\"controlnet `{name}` is not supported for model `{model_name}`.\"\n                )\n        if len(controlnet_model_paths) == 1:\n            kwargs[\"controlnet\"] = controlnet_model_paths[0]\n        else:\n            kwargs[\"controlnet\"] = controlnet_model_paths\n    model_path = cache(model_spec)\n    model = DiffusionModel(\n        model_uid,\n        model_path,\n        lora_model_path=lora_model_path,\n        lora_load_kwargs=lora_load_kwargs,\n        lora_fuse_kwargs=lora_fuse_kwargs,\n        **kwargs,\n    )\n    model_description = ImageModelDescription(\n        subpool_addr, devices, model_spec, model_path=model_path\n    )\n    return model, model_description\n", "path": "xinference/model/image/core.py"}, {"content": "# Copyright 2022-2023 XProbe Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport codecs\nimport json\nimport os\nfrom itertools import chain\n\nfrom .core import (\n    IMAGE_MODEL_DESCRIPTIONS,\n    ImageModelFamilyV1,\n    generate_image_description,\n    get_cache_status,\n    get_image_model_descriptions,\n)\n\n_model_spec_json = os.path.join(os.path.dirname(__file__), \"model_spec.json\")\n_model_spec_modelscope_json = os.path.join(\n    os.path.dirname(__file__), \"model_spec_modelscope.json\"\n)\nBUILTIN_IMAGE_MODELS = dict(\n    (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n    for spec in json.load(codecs.open(_model_spec_json, \"r\", encoding=\"utf-8\"))\n)\nMODELSCOPE_IMAGE_MODELS = dict(\n    (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n    for spec in json.load(\n        codecs.open(_model_spec_modelscope_json, \"r\", encoding=\"utf-8\")\n    )\n)\n\n# register model description\nfor model_name, model_spec in chain(\n    MODELSCOPE_IMAGE_MODELS.items(), BUILTIN_IMAGE_MODELS.items()\n):\n    IMAGE_MODEL_DESCRIPTIONS.update(generate_image_description(model_spec))\n\ndel _model_spec_json\n", "path": "xinference/model/image/__init__.py"}], "after_files": [{"content": "# Copyright 2022-2023 XProbe Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport collections.abc\nimport logging\nimport os\nfrom collections import defaultdict\nfrom typing import Dict, List, Optional, Tuple\n\nfrom ...constants import XINFERENCE_CACHE_DIR\nfrom ..core import CacheableModelSpec, ModelDescription\nfrom ..utils import valid_model_revision\nfrom .stable_diffusion.core import DiffusionModel\n\nMAX_ATTEMPTS = 3\n\nlogger = logging.getLogger(__name__)\n\nIMAGE_MODEL_DESCRIPTIONS: Dict[str, List[Dict]] = defaultdict(list)\nBUILTIN_IMAGE_MODELS: Dict[str, \"ImageModelFamilyV1\"] = {}\nMODELSCOPE_IMAGE_MODELS: Dict[str, \"ImageModelFamilyV1\"] = {}\n\n\ndef get_image_model_descriptions():\n    import copy\n\n    return copy.deepcopy(IMAGE_MODEL_DESCRIPTIONS)\n\n\nclass ImageModelFamilyV1(CacheableModelSpec):\n    model_family: str\n    model_name: str\n    model_id: str\n    model_revision: str\n    model_hub: str = \"huggingface\"\n    controlnet: Optional[List[\"ImageModelFamilyV1\"]]\n\n\nclass ImageModelDescription(ModelDescription):\n    def __init__(\n        self,\n        address: Optional[str],\n        devices: Optional[List[str]],\n        model_spec: ImageModelFamilyV1,\n        model_path: Optional[str] = None,\n    ):\n        super().__init__(address, devices, model_path=model_path)\n        self._model_spec = model_spec\n\n    def to_dict(self):\n        if self._model_spec.controlnet is not None:\n            controlnet = [cn.dict() for cn in self._model_spec.controlnet]\n        else:\n            controlnet = self._model_spec.controlnet\n        return {\n            \"model_type\": \"image\",\n            \"address\": self.address,\n            \"accelerators\": self.devices,\n            \"model_name\": self._model_spec.model_name,\n            \"model_family\": self._model_spec.model_family,\n            \"model_revision\": self._model_spec.model_revision,\n            \"controlnet\": controlnet,\n        }\n\n    def to_version_info(self):\n        from .utils import get_model_version\n\n        if self._model_path is None:\n            is_cached = get_cache_status(self._model_spec)\n            file_location = get_cache_dir(self._model_spec)\n        else:\n            is_cached = True\n            file_location = self._model_path\n\n        if self._model_spec.controlnet is None:\n            return [\n                {\n                    \"model_version\": get_model_version(self._model_spec, None),\n                    \"model_file_location\": file_location,\n                    \"cache_status\": is_cached,\n                    \"controlnet\": \"zoe-depth\",\n                }\n            ]\n        else:\n            res = []\n            for cn in self._model_spec.controlnet:\n                res.append(\n                    {\n                        \"model_version\": get_model_version(self._model_spec, cn),\n                        \"model_file_location\": file_location,\n                        \"cache_status\": is_cached,\n                        \"controlnet\": cn.model_name,\n                    }\n                )\n            return res\n\n\ndef generate_image_description(\n    image_model: ImageModelFamilyV1,\n) -> Dict[str, List[Dict]]:\n    res = defaultdict(list)\n    res[image_model.model_name].extend(\n        ImageModelDescription(None, None, image_model).to_version_info()\n    )\n    return res\n\n\ndef match_diffusion(model_name: str) -> ImageModelFamilyV1:\n    from ..utils import download_from_modelscope\n    from . import BUILTIN_IMAGE_MODELS, MODELSCOPE_IMAGE_MODELS\n\n    if download_from_modelscope():\n        if model_name in MODELSCOPE_IMAGE_MODELS:\n            logger.debug(f\"Image model {model_name} found in ModelScope.\")\n            return MODELSCOPE_IMAGE_MODELS[model_name]\n        else:\n            logger.debug(\n                f\"Image model {model_name} not found in ModelScope, \"\n                f\"now try to load it via builtin way.\"\n            )\n\n    if model_name in BUILTIN_IMAGE_MODELS:\n        return BUILTIN_IMAGE_MODELS[model_name]\n    else:\n        raise ValueError(\n            f\"Image model {model_name} not found, available\"\n            f\"model list: {BUILTIN_IMAGE_MODELS.keys()}\"\n        )\n\n\ndef cache(model_spec: ImageModelFamilyV1):\n    from ..utils import cache\n\n    return cache(model_spec, ImageModelDescription)\n\n\ndef get_cache_dir(model_spec: ImageModelFamilyV1):\n    return os.path.realpath(os.path.join(XINFERENCE_CACHE_DIR, model_spec.model_name))\n\n\ndef get_cache_status(\n    model_spec: ImageModelFamilyV1,\n) -> bool:\n    cache_dir = get_cache_dir(model_spec)\n    meta_path = os.path.join(cache_dir, \"__valid_download\")\n\n    model_name = model_spec.model_name\n    if model_name in BUILTIN_IMAGE_MODELS and model_name in MODELSCOPE_IMAGE_MODELS:\n        hf_spec = BUILTIN_IMAGE_MODELS[model_name]\n        ms_spec = MODELSCOPE_IMAGE_MODELS[model_name]\n\n        return any(\n            [\n                valid_model_revision(meta_path, hf_spec.model_revision),\n                valid_model_revision(meta_path, ms_spec.model_revision),\n            ]\n        )\n    else:  # Usually for UT\n        logger.warning(f\"Cannot find builtin image model spec: {model_name}\")\n        return valid_model_revision(meta_path, model_spec.model_revision)\n\n\ndef create_image_model_instance(\n    subpool_addr: str,\n    devices: List[str],\n    model_uid: str,\n    model_name: str,\n    lora_model_path: Optional[str] = None,\n    lora_load_kwargs: Optional[Dict] = None,\n    lora_fuse_kwargs: Optional[Dict] = None,\n    **kwargs,\n) -> Tuple[DiffusionModel, ImageModelDescription]:\n    model_spec = match_diffusion(model_name)\n    controlnet = kwargs.get(\"controlnet\")\n    # Handle controlnet\n    if controlnet is not None:\n        if isinstance(controlnet, str):\n            controlnet = [controlnet]\n        elif not isinstance(controlnet, collections.abc.Sequence):\n            raise ValueError(\"controlnet should be a str or a list of str.\")\n        elif set(controlnet) != len(controlnet):\n            raise ValueError(\"controlnet should be a list of unique str.\")\n        elif not model_spec.controlnet:\n            raise ValueError(f\"Model {model_name} has empty controlnet list.\")\n\n        controlnet_model_paths = []\n        assert model_spec.controlnet is not None\n        for name in controlnet:\n            for cn_model_spec in model_spec.controlnet:\n                if cn_model_spec.model_name == name:\n                    model_path = cache(cn_model_spec)\n                    controlnet_model_paths.append(model_path)\n                    break\n            else:\n                raise ValueError(\n                    f\"controlnet `{name}` is not supported for model `{model_name}`.\"\n                )\n        if len(controlnet_model_paths) == 1:\n            kwargs[\"controlnet\"] = controlnet_model_paths[0]\n        else:\n            kwargs[\"controlnet\"] = controlnet_model_paths\n    model_path = cache(model_spec)\n    model = DiffusionModel(\n        model_uid,\n        model_path,\n        lora_model_path=lora_model_path,\n        lora_load_kwargs=lora_load_kwargs,\n        lora_fuse_kwargs=lora_fuse_kwargs,\n        **kwargs,\n    )\n    model_description = ImageModelDescription(\n        subpool_addr, devices, model_spec, model_path=model_path\n    )\n    return model, model_description\n", "path": "xinference/model/image/core.py"}, {"content": "# Copyright 2022-2023 XProbe Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport codecs\nimport json\nimport os\nfrom itertools import chain\n\nfrom .core import (\n    BUILTIN_IMAGE_MODELS,\n    IMAGE_MODEL_DESCRIPTIONS,\n    MODELSCOPE_IMAGE_MODELS,\n    ImageModelFamilyV1,\n    generate_image_description,\n    get_cache_status,\n    get_image_model_descriptions,\n)\n\n_model_spec_json = os.path.join(os.path.dirname(__file__), \"model_spec.json\")\n_model_spec_modelscope_json = os.path.join(\n    os.path.dirname(__file__), \"model_spec_modelscope.json\"\n)\nBUILTIN_IMAGE_MODELS.update(\n    dict(\n        (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n        for spec in json.load(codecs.open(_model_spec_json, \"r\", encoding=\"utf-8\"))\n    )\n)\nMODELSCOPE_IMAGE_MODELS.update(\n    dict(\n        (spec[\"model_name\"], ImageModelFamilyV1(**spec))\n        for spec in json.load(\n            codecs.open(_model_spec_modelscope_json, \"r\", encoding=\"utf-8\")\n        )\n    )\n)\n\n# register model description\nfor model_name, model_spec in chain(\n    MODELSCOPE_IMAGE_MODELS.items(), BUILTIN_IMAGE_MODELS.items()\n):\n    IMAGE_MODEL_DESCRIPTIONS.update(generate_image_description(model_spec))\n\ndel _model_spec_json\n", "path": "xinference/model/image/__init__.py"}]}