alphard/ComfyUI-Lora-Manager
1
0
Fork 0
mirror of https://github.com/willmiao/ComfyUI-Lora-Manager.git synced 2026-08-08 23:10:15 -03:00
Code Issues Packages Projects Releases Wiki Activity

fix(types): resolve pre-existing basedpyright errors in py/ and standalone.py

Browse Source 
Fix ~950 basedpyright errors across the backend:
- Convert ineffective # type: ignore comments to # pyright: ignore[rule]
- Add missing generic type arguments (Dict[str, Any], list[Any], ...)
- Annotate dynamic dict literals and runtime-initialized attributes
- Widen CivitAI provider tuple signatures in recipe parsers
- Remove dead LoraRoutes handlers calling nonexistent LoraService methods
- Suppress unavoidable ServiceRegistry import cycles (basedpyright counts
  function-local imports as cycle edges)
This commit is contained in:
Will Miao
2026-08-08 20:12:52 +08:00
parent 6fcdeb799d
commit 8e724538bd
103 changed files with 1184 additions and 1015 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
py/utils/civitai_utils.py
View File

@@ -224,7 +224,7 @@ def _normalize_commercial_values(value: Any) -> Sequence[str]:
if result:
return result
try:
if len(value) == 0: # type: ignore[arg-type]
if len(value) == 0: # pyright: ignore[reportArgumentType]
return []
except TypeError:
pass

24
py/utils/example_images_download_manager.py
View File

@@ -35,7 +35,7 @@ class ExampleImagesDownloadError(RuntimeError):
class DownloadInProgressError(ExampleImagesDownloadError):
"""Raised when a download is already running."""
def __init__(self, progress_snapshot: dict) -> None:
def __init__(self, progress_snapshot: Dict[str, Any]) -> None:
super().__init__("Download already in progress")
self.progress_snapshot = progress_snapshot
@@ -54,7 +54,7 @@ class DownloadConfigurationError(ExampleImagesDownloadError):
logger = logging.getLogger(__name__)
class _DownloadProgress(dict):
class _DownloadProgress(dict[str, Any]):
"""Mutable mapping maintaining download progress with set-aware serialisation."""
def __init__(self) -> None:
@@ -80,7 +80,7 @@ class _DownloadProgress(dict):
rate_limited_models=set(),
)
def snapshot(self) -> dict:
def snapshot(self) -> Dict[str, Any]:
"""Return a JSON-serialisable snapshot of the current progress."""
snapshot = dict(self)
@@ -149,7 +149,7 @@ class DownloadManager:
"""Manages downloading example images for models."""
def __init__(self, *, ws_manager, state_lock: asyncio.Lock | None = None) -> None:
self._download_task: asyncio.Task | None = None
self._download_task: asyncio.Task[Any] | None = None
self._is_downloading = False
self._progress = _DownloadProgress()
self._ws_manager = ws_manager
@@ -162,7 +162,7 @@ class DownloadManager:
return ""
return ensure_library_root_exists(library_name)
async def start_download(self, options: dict):
async def start_download(self, options: Dict[str, Any]):
"""Start downloading example images for models."""
# Step 1: Parse options (fast, non-blocking)
@@ -269,7 +269,7 @@ class DownloadManager:
return {"success": True, "message": "Download started", "status": snapshot}
def _handle_download_task_done(self, task: asyncio.Task, output_dir: str) -> None:
def _handle_download_task_done(self, task: asyncio.Task[Any], output_dir: str) -> None:
"""Handle download task completion, including saving progress on error."""
try:
# This will re-raise any exception from the task
@@ -282,7 +282,7 @@ class DownloadManager:
except Exception as save_error:
logger.error(f"Failed to save progress after task failure: {save_error}")
async def get_status(self, request) -> dict:
async def get_status(self, request) -> Dict[str, Any]:
"""Get the current status of example images download."""
return {
@@ -291,7 +291,7 @@ class DownloadManager:
"status": self._progress.snapshot(),
}
async def _load_progress_file(self, output_dir: str) -> tuple[str, set, set, set]:
async def _load_progress_file(self, output_dir: str) -> tuple[str, set[str], set[str], set[str]]:
"""Load progress file from disk. Returns (progress_file_path, processed_models, failed_models, rate_limited_models).
This is a separate async method to allow running in executor to avoid blocking event loop.
@@ -301,7 +301,7 @@ class DownloadManager:
None, self._load_progress_file_sync, output_dir
)
def _load_progress_file_sync(self, output_dir: str) -> tuple[str, set, set, set]:
def _load_progress_file_sync(self, output_dir: str) -> tuple[str, set[str], set[str], set[str]]:
"""Synchronous implementation of progress file loading.
Returns:
@@ -356,7 +356,7 @@ class DownloadManager:
return progress_file, processed_models, failed_models, rate_limited_models
def _load_progress_sets_sync(self, progress_file: str) -> tuple[set, set]:
def _load_progress_sets_sync(self, progress_file: str) -> tuple[set[str], set[str]]:
"""Load only the processed and failed model sets from progress file.
This is a lighter version for quick checks without legacy migration.
@@ -377,7 +377,7 @@ class DownloadManager:
return processed_models, failed_models
async def check_pending_models(self, model_types: list[str]) -> dict:
async def check_pending_models(self, model_types: list[str]) -> Dict[str, Any]:
"""Quickly check how many models need example images downloaded.
This is a lightweight check that avoids the overhead of starting
@@ -1000,7 +1000,7 @@ class DownloadManager:
except Exception as e:
logger.error(f"Failed to save progress file: {e}")
async def start_force_download(self, options: dict):
async def start_force_download(self, options: Dict[str, Any]):
"""Force download example images for specific models."""
async with self._state_lock:

22
py/utils/example_images_metadata.py
View File

@@ -93,7 +93,7 @@ class MetadataUpdater:
"""Handles updating model metadata related to example images"""
@staticmethod
async def refresh_model_metadata(model_hash, model_name, scanner_type, scanner, progress: dict | None = None):
async def refresh_model_metadata(model_hash, model_name, scanner_type, scanner, progress: dict[str, Any] | None = None):
"""Refresh model metadata from CivitAI
Args:
@@ -263,8 +263,9 @@ class MetadataUpdater:
model_copy: Optional[Dict[str, Any]] = None
try:
model_copy = model.copy()
model_copy.pop('folder', None)
await MetadataManager.save_metadata(file_path, model_copy)
if model_copy is not None:
model_copy.pop('folder', None)
await MetadataManager.save_metadata(file_path, model_copy)
logger.info(f"Saved metadata for {model.get('model_name')}")
except Exception as e:
logger.error(f"Failed to save metadata for {model.get('model_name')}: {str(e)}")
@@ -371,8 +372,9 @@ class MetadataUpdater:
if file_path:
try:
model_copy = model_data.copy()
model_copy.pop('folder', None)
await MetadataManager.save_metadata(file_path, model_copy)
if model_copy is not None:
model_copy.pop('folder', None)
await MetadataManager.save_metadata(file_path, model_copy)
logger.info(f"Saved metadata for {model_data.get('model_name')}")
except Exception as e:
logger.error(f"Failed to save metadata: {str(e)}")
@@ -553,7 +555,7 @@ class MetadataUpdater:
images = civitai.get("images")
if isinstance(images, list) and images:
stale: list[int] = []
stale_images: list[int] = []
for idx, img in enumerate(images):
if img.get("url", ""):
@@ -563,15 +565,15 @@ class MetadataUpdater:
prefix = f"image_{idx}."
if not any(f.startswith(prefix) for f in dir_entries):
stale.append(idx)
stale_images.append(idx)
if stale:
for idx in reversed(stale):
if stale_images:
for idx in reversed(stale_images):
images.pop(idx)
has_changes = True
logger.info(
"Pruned %d stale image entry(ies) for %s",
len(stale),
len(stale_images),
getattr(metadata, "model_name", model_hash),
)

2
py/utils/example_images_migration.py
View File

@@ -371,7 +371,7 @@ class ExampleImagesMigration:
found = True
break
if not found:
if not found or old_path is None:
logger.warning(f"Could not find file for index {index} in {model_hash}, skipping")
continue

3
py/utils/example_images_processor.py
View File

@@ -4,6 +4,7 @@ import os
import re
import random
import string
from typing import Any
from aiohttp import web
from ..utils.constants import SUPPORTED_MEDIA_EXTENSIONS
from ..services.service_registry import ServiceRegistry
@@ -259,7 +260,7 @@ class ExampleImagesProcessor:
logger.debug("File already exists, skipping download for %s", image_url)
continue
async def _attempt_download() -> tuple:
async def _attempt_download() -> tuple[bool, Any, Any]:
logger.debug("Downloading media file %s for %s", i, model_name)
return await downloader.download_to_memory(
image_url,

52
py/utils/exif_utils.py
View File

@@ -4,13 +4,13 @@ import logging
import os
import struct
from io import BytesIO
from typing import Any, Optional, Tuple
from typing import Any, Optional, Tuple, cast
import piexif
import piexif # pyright: ignore[reportMissingTypeStubs]
from PIL import Image, PngImagePlugin
try:
import brotli
import brotli # pyright: ignore[reportMissingTypeStubs]
_BROTLI_AVAILABLE = True
except ImportError:
brotli = None
@@ -38,7 +38,7 @@ class ExifUtils:
"""Utility functions for working with EXIF data in images"""
@staticmethod
def _parse_isobmff_boxes(data: bytes, offset: int = 0) -> list[dict]:
def _parse_isobmff_boxes(data: bytes, offset: int = 0) -> list[dict[str, Any]]:
boxes = []
while offset + 8 <= len(data):
size = struct.unpack('>I', data[offset:offset + 4])[0]
@@ -78,7 +78,7 @@ class ExifUtils:
_BROTLI_MAX_DECOMPRESSED = 2 * 1024 * 1024
@staticmethod
def _extract_isobmff_brotli(image_path: str) -> Optional[dict]:
def _extract_isobmff_brotli(image_path: str) -> Optional[dict[str, Any]]:
try:
with open(image_path, 'rb') as f:
data = f.read()
@@ -107,7 +107,7 @@ class ExifUtils:
if _BROTLI_AVAILABLE:
try:
decompressed = brotli.decompress(compressed)
decompressed = brotli.decompress(compressed) # pyright: ignore[reportOptionalMemberAccess]
if len(decompressed) > ExifUtils._BROTLI_MAX_DECOMPRESSED:
logger.warning(
"Brotli metadata too large (%d bytes, max %d), ignoring",
@@ -126,7 +126,9 @@ class ExifUtils:
except Exception:
return None
result = {"parameters": None, "prompt": None, "workflow": None, "comment": None}
result: dict[str, Optional[str]] = {
"parameters": None, "prompt": None, "workflow": None, "comment": None
}
if isinstance(meta.get("prompt"), (dict, list)):
result["prompt"] = json.dumps(meta["prompt"])
elif isinstance(meta.get("prompt"), str):
@@ -161,7 +163,7 @@ class ExifUtils:
@staticmethod
def _load_structured_metadata(image_path: str) -> dict[str, Optional[str]]:
metadata = {
metadata: dict[str, Optional[str]] = {
"parameters": None,
"prompt": None,
"workflow": None,
@@ -197,13 +199,14 @@ class ExifUtils:
logger.debug(f"Error loading EXIF data: {e}")
exif_dict = {}
if piexif.ExifIFD.UserComment in exif_dict.get("Exif", {}):
exif_ifd = exif_dict.get("Exif")
if exif_ifd and piexif.ExifIFD.UserComment in exif_ifd:
metadata["comment"] = ExifUtils._decode_user_comment(
exif_dict["Exif"][piexif.ExifIFD.UserComment]
exif_ifd[piexif.ExifIFD.UserComment]
)
image_description = ExifUtils._decode_exif_text(
exif_dict.get("0th", {}).get(piexif.ImageIFD.ImageDescription)
(exif_dict.get("0th") or {}).get(piexif.ImageIFD.ImageDescription)
)
if image_description:
if image_description.startswith("Workflow:"):
@@ -253,19 +256,26 @@ class ExifUtils:
workflow = metadata_fields.get("workflow")
prompt = metadata_fields.get("prompt")
# Work on local references, then write the (possibly new) IFD dicts back.
exif_ifd = exif_dict.get("Exif") or {}
exif_0th = exif_dict.get("0th") or {}
if parameters:
exif_dict["Exif"][piexif.ExifIFD.UserComment] = (
exif_ifd[piexif.ExifIFD.UserComment] = (
b"UNICODE\0" + parameters.encode("utf-16be")
)
else:
exif_dict["Exif"].pop(piexif.ExifIFD.UserComment, None)
exif_ifd.pop(piexif.ExifIFD.UserComment, None)
if workflow:
exif_dict["0th"][piexif.ImageIFD.ImageDescription] = f"Workflow:{workflow}"
exif_0th[piexif.ImageIFD.ImageDescription] = f"Workflow:{workflow}"
elif prompt:
exif_dict["0th"][piexif.ImageIFD.ImageDescription] = prompt
exif_0th[piexif.ImageIFD.ImageDescription] = prompt
else:
exif_dict["0th"].pop(piexif.ImageIFD.ImageDescription, None)
exif_0th.pop(piexif.ImageIFD.ImageDescription, None)
exif_dict["Exif"] = exif_ifd
exif_dict["0th"] = exif_0th
return piexif.dump(exif_dict)
@@ -326,7 +336,7 @@ class ExifUtils:
exif_bytes = ExifUtils._build_exif_bytes(
metadata_fields, img.info.get("exif")
)
save_kwargs = {"exif": exif_bytes}
save_kwargs: dict[str, Any] = {"exif": exif_bytes}
if img_format == "WEBP":
save_kwargs["quality"] = 85
@@ -499,12 +509,12 @@ class ExifUtils:
else:
# It's binary data - validate data
try:
with BytesIO(image_data) as temp_buf:
with BytesIO(cast(bytes, image_data)) as temp_buf:
test_img = Image.open(temp_buf)
# Verify the image can be fully loaded
width, height = test_img.size
# If successful, reopen for processing
img = Image.open(BytesIO(image_data))
img = Image.open(BytesIO(cast(bytes, image_data)))
except Exception as e:
logger.error(f"Invalid binary image data: {e}")
raise ValueError(f"Cannot process corrupt image data: {e}")
@@ -521,7 +531,7 @@ class ExifUtils:
import tempfile
with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as temp_file:
temp_path = temp_file.name
temp_file.write(image_data)
temp_file.write(cast(bytes, image_data))
try:
metadata_fields = ExifUtils._load_structured_metadata(temp_path)
except Exception as e:
@@ -542,7 +552,7 @@ class ExifUtils:
# Resize the image with error handling
try:
resized_img = img.resize((target_width, new_height), Image.LANCZOS)
resized_img = img.resize((target_width, new_height), Image.Resampling.LANCZOS)
except Exception as e:
logger.error(f"Failed to resize image: {e}")
# Return original image if resize fails

8
py/utils/lora_metadata.py
View File

@@ -1,5 +1,5 @@
from safetensors import safe_open
from typing import Dict, List, Tuple
from typing import Dict, List, Optional, Tuple
from .model_utils import determine_base_model
import os
import logging
@@ -7,7 +7,7 @@ import json
logger = logging.getLogger(__name__)
async def extract_lora_metadata(file_path: str) -> Dict:
async def extract_lora_metadata(file_path: str) -> Dict[str, str]:
"""Extract essential metadata from safetensors file"""
try:
with safe_open(file_path, framework="pt", device="cpu") as f:
@@ -20,7 +20,7 @@ async def extract_lora_metadata(file_path: str) -> Dict:
logger.error(f"Error reading metadata from {file_path}: {str(e)}")
return {"base_model": "Unknown"}
async def extract_checkpoint_metadata(file_path: str) -> dict:
async def extract_checkpoint_metadata(file_path: str) -> dict[str, str]:
"""Extract metadata from a checkpoint file to determine model type and base model"""
try:
# Analyze filename for clues about the model
@@ -83,7 +83,7 @@ async def extract_checkpoint_metadata(file_path: str) -> dict:
# Return default values
return {'base_model': 'Unknown', 'model_type': 'checkpoint'}
async def extract_trained_words(file_path: str) -> Tuple[List[Tuple[str, int]], str]:
async def extract_trained_words(file_path: str) -> Tuple[List[Tuple[str, int]], Optional[str]]:
"""Extract trained words from a safetensors file and sort by frequency
Args:

16
py/utils/metadata_manager.py
View File

@@ -3,9 +3,9 @@ import os
import json
import logging
import time
from typing import Any, Dict, Optional, Type, Union
from typing import Any, Dict, Optional, Type, Union, cast
from .models import BaseModelMetadata, LoraMetadata
from .models import BaseModelMetadata, CheckpointMetadata, EmbeddingMetadata, LoraMetadata
from .file_utils import normalize_path, find_preview_file, calculate_sha256, calculate_autov3
from .lora_metadata import extract_lora_metadata, extract_checkpoint_metadata
@@ -56,13 +56,13 @@ class MetadataManager:
return None, True # should_skip = True
@staticmethod
async def load_metadata_payload(file_path: str) -> Dict:
async def load_metadata_payload(file_path: str) -> Dict[str, Any]:
"""
Load metadata and return it as a dictionary, including any unknown fields.
Falls back to reading the raw JSON file if parsing into a model class fails.
"""
payload: Dict = {}
payload: Dict[str, Any] = {}
metadata_obj, should_skip = await MetadataManager.load_metadata(file_path)
if metadata_obj:
@@ -120,7 +120,7 @@ class MetadataManager:
return model_data
@staticmethod
async def save_metadata(path: str, metadata: Union[BaseModelMetadata, Dict]) -> bool:
async def save_metadata(path: str, metadata: Union[BaseModelMetadata, Dict[str, Any]]) -> bool:
"""
Save metadata with atomic write operations.
@@ -217,7 +217,7 @@ class MetadataManager:
# Create instance based on model type
if model_class.__name__ == "CheckpointMetadata":
metadata = model_class(
metadata = cast(Type[CheckpointMetadata], model_class)(
file_name=base_name,
model_name=base_name,
file_path=normalize_path(file_path),
@@ -232,7 +232,7 @@ class MetadataManager:
from_civitai=True
)
elif model_class.__name__ == "EmbeddingMetadata":
metadata = model_class(
metadata = cast(Type[EmbeddingMetadata], model_class)(
file_name=base_name,
model_name=base_name,
file_path=normalize_path(file_path),
@@ -247,7 +247,7 @@ class MetadataManager:
from_civitai=True
)
else: # Default to LoraMetadata
metadata = model_class(
metadata = cast(Type[LoraMetadata], model_class)(
file_name=base_name,
model_name=base_name,
file_path=normalize_path(file_path),

20
py/utils/models.py
View File

@@ -1,5 +1,5 @@
from dataclasses import dataclass, asdict, field
from typing import Dict, Optional, List, Any
from typing import Callable, Dict, Optional, List, Any
from datetime import datetime
import os
from .constants import INVALID_AUTOV3_EMPTY_HASH
@@ -23,7 +23,7 @@ def normalize_autov3(value: Any) -> Optional[str]:
return None
def autov3_from_civitai_files(civitai_data: Optional[Dict], sha256: str) -> Optional[str]:
def autov3_from_civitai_files(civitai_data: Optional[Dict[str, Any]], sha256: str) -> Optional[str]:
"""Extract the AutoV3 hash from Civitai metadata for the matching file.
Civitai versions can ship multiple files; the AutoV3 hash is only valid
@@ -64,7 +64,7 @@ class BaseModelMetadata:
civitai: Dict[str, Any] = field(
default_factory=dict
) # Civitai API data if available
tags: List[str] = None # Model tags
tags: List[str] = field(default_factory=list) # Model tags
modelDescription: str = "" # Full model description
civitai_deleted: bool = False # Whether deleted from Civitai
favorite: bool = False # Whether the model is a favorite
@@ -96,7 +96,7 @@ class BaseModelMetadata:
self.trainedWords = []
@classmethod
def from_dict(cls, data: Dict) -> "BaseModelMetadata":
def from_dict(cls, data: Dict[str, Any]) -> "BaseModelMetadata":
"""Create instance from dictionary"""
data_copy = data.copy()
@@ -136,7 +136,7 @@ class BaseModelMetadata:
return instance
def to_dict(self) -> Dict:
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization"""
result = asdict(self)
@@ -158,7 +158,7 @@ class BaseModelMetadata:
return result
def update_civitai_info(self, civitai_data: Dict) -> None:
def update_civitai_info(self, civitai_data: Dict[str, Any]) -> None:
"""Update Civitai information.
Civitai's AutoV3 is the authoritative hash for recipe matching, so
@@ -192,7 +192,7 @@ class BaseModelMetadata:
@staticmethod
def generate_unique_filename(
target_dir: str, base_name: str, extension: str, hash_provider: callable = None
target_dir: str, base_name: str, extension: str, hash_provider: Optional[Callable[[], str]] = None
) -> str:
"""Generate a unique filename to avoid conflicts
@@ -243,7 +243,7 @@ class LoraMetadata(BaseModelMetadata):
@classmethod
def from_civitai_info(
cls, version_info: Dict, file_info: Dict, save_path: str
cls, version_info: Dict[str, Any], file_info: Dict[str, Any], save_path: str
) -> "LoraMetadata":
"""Create LoraMetadata instance from Civitai version info"""
file_name = file_info.get("name", "")
@@ -287,7 +287,7 @@ class CheckpointMetadata(BaseModelMetadata):
@classmethod
def from_civitai_info(
cls, version_info: Dict, file_info: Dict, save_path: str
cls, version_info: Dict[str, Any], file_info: Dict[str, Any], save_path: str
) -> "CheckpointMetadata":
"""Create CheckpointMetadata instance from Civitai version info"""
file_name = file_info.get("name", "")
@@ -332,7 +332,7 @@ class EmbeddingMetadata(BaseModelMetadata):
@classmethod
def from_civitai_info(
cls, version_info: Dict, file_info: Dict, save_path: str
cls, version_info: Dict[str, Any], file_info: Dict[str, Any], save_path: str
) -> "EmbeddingMetadata":
"""Create EmbeddingMetadata instance from Civitai version info"""
file_name = file_info.get("name", "")

2
py/utils/preview_selection.py
View File

@@ -15,7 +15,7 @@ def _extract_nsfw_level(entry: Mapping[str, object]) -> int:
value = entry.get("nsfwLevel", 0)
try:
return int(value) # type: ignore[return-value]
return int(value) # pyright: ignore[reportArgumentType, reportReturnType]
except (TypeError, ValueError):
return 0

20
py/utils/usage_stats.py
View File

@@ -6,7 +6,7 @@ import asyncio
import logging
import datetime
import shutil
from typing import Dict, Set
from typing import Any, Awaitable, Dict, Set, cast
from ..config import config
from ..services.service_registry import ServiceRegistry
@@ -68,7 +68,7 @@ class UsageStats:
return
# Initialize stats storage
self.stats = {
self.stats: Dict[str, Any] = {
"checkpoints": {}, # sha256 -> { total: count, history: { date: count } }
"loras": {}, # sha256 -> { total: count, history: { date: count } }
"embeddings": {}, # sha256 -> { total: count, history: { date: count } }
@@ -297,8 +297,8 @@ class UsageStats:
# Process each prompt_id
try:
registry = MetadataRegistry()
except NameError:
registry = MetadataRegistry() # pyright: ignore[reportPossiblyUnboundVariable]
except (ImportError, NameError):
# MetadataRegistry not available (standalone mode)
registry = None
@@ -374,7 +374,7 @@ class UsageStats:
if not callable(get_cached_data):
return None
cache = await get_cached_data()
cache = await cast(Awaitable[Any], get_cached_data())
raw_data = getattr(cache, "raw_data", None)
if not isinstance(raw_data, list):
return None
@@ -404,7 +404,7 @@ class UsageStats:
if not callable(get_model_roots):
return None
roots = [root for root in get_model_roots() if root]
roots = [root for root in cast(Any, get_model_roots()) if root]
if not roots:
return None
@@ -486,7 +486,7 @@ class UsageStats:
model_filename,
file_path,
)
calculated_hash = await calculate_hash(file_path)
calculated_hash = await cast(Awaitable[Any], calculate_hash(file_path))
if calculated_hash:
return calculated_hash
@@ -557,7 +557,7 @@ class UsageStats:
logger.error(f"Error processing LoRA usage: {e}", exc_info=True)
@staticmethod
def _extract_embedding_names(prompt_text: str) -> set:
def _extract_embedding_names(prompt_text: str) -> set[str]:
"""Parse embedding:name references from prompt text.
ComfyUI's SDTokenizer resolves ``embedding:<name>`` during tokenization
@@ -605,7 +605,7 @@ class UsageStats:
except Exception as e:
logger.error("Error processing embedding usage: %s", e, exc_info=True)
async def get_stats(self):
async def get_stats(self) -> Dict[str, Any]:
"""Get current usage statistics"""
return self.stats
@@ -633,7 +633,7 @@ class UsageStats:
try:
# Process metadata for this prompt_id
registry = MetadataRegistry()
registry = MetadataRegistry() # pyright: ignore[reportPossiblyUnboundVariable]
metadata = registry.get_metadata(prompt_id)
if metadata:
await self._process_metadata(metadata)

4
py/utils/utils.py
View File

@@ -261,7 +261,7 @@ def get_checkpoint_info_absolute(checkpoint_name):
return asyncio.run(_get_checkpoint_info_absolute_async())
def _format_model_name_for_comfyui(file_path: str, model_roots: list) -> str:
def _format_model_name_for_comfyui(file_path: str, model_roots: list[str]) -> str:
"""Format file path to ComfyUI-style model name (relative path with extension)
Example: /path/to/checkpoints/Illustrious/model.safetensors -> Illustrious/model.safetensors
@@ -470,7 +470,7 @@ def calculate_recipe_fingerprint(loras):
def calculate_relative_path_for_model(
model_data: Dict, model_type: str = "lora"
model_data: Dict[str, Any], model_type: str = "lora"
) -> str:
"""Calculate relative path for existing model using template from settings