110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
# LoRA catalog — curated, scope-filtered LoRA packs the model panel offers as
|
|
# swappable "styles". Data lives in data/loras/*.json (shipped) and
|
|
# user/PromptChain/loras/*.json (user-added, shadows by id). Each file:
|
|
# { "scope": {type, architecture, family}, "loras": [ {id,label,filename,
|
|
# folder,trigger,strength_default,size_bytes,source}, ... ] }
|
|
# Served scope-filtered to the active model so e.g. Krea 2 Turbo styles only
|
|
# surface on Krea 2 Turbo.
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from aiohttp import web
|
|
import server
|
|
import folder_paths
|
|
|
|
routes = server.PromptServer.instance.routes
|
|
|
|
_cache: Optional[list] = None
|
|
_cache_signature: Optional[tuple] = None
|
|
|
|
|
|
def _system_dir() -> Path:
|
|
return Path(__file__).parent.parent / "data" / "loras"
|
|
|
|
|
|
def _user_dir() -> Path:
|
|
return Path(folder_paths.get_user_directory()) / "PromptChain" / "loras"
|
|
|
|
|
|
def _dir_signature() -> tuple:
|
|
# Mirror model_settings' mtime watch: a dir's mtime bumps when files are
|
|
# added/removed/atomically replaced (a /deploy pull), so shipped catalogs
|
|
# refresh without a restart. In-place edits to an existing file don't bump
|
|
# the dir mtime — those need a restart.
|
|
sig = []
|
|
for directory in (_system_dir(), _user_dir()):
|
|
try:
|
|
sig.append(directory.stat().st_mtime_ns)
|
|
except OSError:
|
|
sig.append(0)
|
|
return tuple(sig)
|
|
|
|
|
|
def _read_json(path: Path) -> Optional[dict]:
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
return data if isinstance(data, dict) else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _ingest() -> list:
|
|
"""All catalog files as (scope, loras) records. User dir is read last so a
|
|
user catalog with the same lora id shadows the shipped one."""
|
|
by_id: dict[str, dict] = {}
|
|
order: list[str] = []
|
|
for directory in (_system_dir(), _user_dir()):
|
|
if not directory.is_dir():
|
|
continue
|
|
for path in sorted(directory.glob("*.json")):
|
|
data = _read_json(path)
|
|
if not data:
|
|
continue
|
|
scope = data.get("scope", {})
|
|
for lora in data.get("loras", []):
|
|
lid = lora.get("id") or lora.get("filename")
|
|
if not lid:
|
|
continue
|
|
if lid not in by_id:
|
|
order.append(lid)
|
|
by_id[lid] = {**lora, "scope": scope}
|
|
return [by_id[lid] for lid in order]
|
|
|
|
|
|
def _get_all() -> list:
|
|
global _cache, _cache_signature
|
|
sig = _dir_signature()
|
|
if _cache is not None and sig == _cache_signature:
|
|
return _cache
|
|
_cache = _ingest()
|
|
_cache_signature = sig
|
|
return _cache
|
|
|
|
|
|
def _matches_scope(scope: dict, architecture: str, family: Optional[str]) -> bool:
|
|
scope_type = scope.get("type", "architecture")
|
|
if scope_type == "family":
|
|
return scope.get("architecture") == architecture and scope.get("family") == family
|
|
return scope.get("architecture") == architecture
|
|
|
|
|
|
def list_loras(architecture: str, family: Optional[str] = None) -> list:
|
|
"""Scope-filtered LoRAs for the active model, scope stripped from each."""
|
|
out = []
|
|
for lora in _get_all():
|
|
if _matches_scope(lora.get("scope", {}), architecture, family):
|
|
out.append({k: v for k, v in lora.items() if k != "scope"})
|
|
return out
|
|
|
|
|
|
@routes.get("/promptchain/loras/catalog")
|
|
async def _api_lora_catalog(request):
|
|
architecture = request.query.get("architecture", "").strip()
|
|
family = request.query.get("family", "").strip() or None
|
|
if not architecture:
|
|
return web.json_response({"loras": []})
|
|
return web.json_response({"loras": list_loras(architecture, family)})
|