mirror of
https://github.com/invoke-ai/InvokeAI
synced 2024-08-30 20:32:17 +00:00
922716d2ab
The previous super-minimal implementation had a major issue - the saved workflow didn't take into account batched field values. When generating with multiple iterations or dynamic prompts, the same workflow with the first prompt, seed, etc was stored in each image. As a result, when the batch results in multiple queue items, only one of the images has the correct workflow - the others are mismatched. To work around this, we can store the _graph_ in the image metadata (alongside the workflow, if generated via workflow editor). When loading a workflow from an image, we can choose to load the workflow or the graph, preferring the workflow. Internally, we need to update images router image-saving services. The changes are minimal. To avoid pydantic errors deserializing the graph, when we extract it from the image, we will leave it as stringified JSON and let the frontend's more sophisticated and flexible parsing handle it. The worklow is also changed to just return stringified JSON, so the API is consistent.
148 lines
4.3 KiB
Python
148 lines
4.3 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Callable, Optional
|
|
|
|
from PIL.Image import Image as PILImageType
|
|
|
|
from invokeai.app.invocations.fields import MetadataField
|
|
from invokeai.app.services.image_records.image_records_common import (
|
|
ImageCategory,
|
|
ImageRecord,
|
|
ImageRecordChanges,
|
|
ResourceOrigin,
|
|
)
|
|
from invokeai.app.services.images.images_common import ImageDTO
|
|
from invokeai.app.services.shared.graph import Graph
|
|
from invokeai.app.services.shared.pagination import OffsetPaginatedResults
|
|
from invokeai.app.services.workflow_records.workflow_records_common import WorkflowWithoutID
|
|
|
|
|
|
class ImageServiceABC(ABC):
|
|
"""High-level service for image management."""
|
|
|
|
_on_changed_callbacks: list[Callable[[ImageDTO], None]]
|
|
_on_deleted_callbacks: list[Callable[[str], None]]
|
|
|
|
def __init__(self) -> None:
|
|
self._on_changed_callbacks = []
|
|
self._on_deleted_callbacks = []
|
|
|
|
def on_changed(self, on_changed: Callable[[ImageDTO], None]) -> None:
|
|
"""Register a callback for when an image is changed"""
|
|
self._on_changed_callbacks.append(on_changed)
|
|
|
|
def on_deleted(self, on_deleted: Callable[[str], None]) -> None:
|
|
"""Register a callback for when an image is deleted"""
|
|
self._on_deleted_callbacks.append(on_deleted)
|
|
|
|
def _on_changed(self, item: ImageDTO) -> None:
|
|
for callback in self._on_changed_callbacks:
|
|
callback(item)
|
|
|
|
def _on_deleted(self, item_id: str) -> None:
|
|
for callback in self._on_deleted_callbacks:
|
|
callback(item_id)
|
|
|
|
@abstractmethod
|
|
def create(
|
|
self,
|
|
image: PILImageType,
|
|
image_origin: ResourceOrigin,
|
|
image_category: ImageCategory,
|
|
node_id: Optional[str] = None,
|
|
session_id: Optional[str] = None,
|
|
board_id: Optional[str] = None,
|
|
is_intermediate: Optional[bool] = False,
|
|
metadata: Optional[MetadataField] = None,
|
|
workflow: Optional[WorkflowWithoutID] = None,
|
|
graph: Optional[Graph] = None,
|
|
) -> ImageDTO:
|
|
"""Creates an image, storing the file and its metadata."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def update(
|
|
self,
|
|
image_name: str,
|
|
changes: ImageRecordChanges,
|
|
) -> ImageDTO:
|
|
"""Updates an image."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_pil_image(self, image_name: str) -> PILImageType:
|
|
"""Gets an image as a PIL image."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_record(self, image_name: str) -> ImageRecord:
|
|
"""Gets an image record."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_dto(self, image_name: str) -> ImageDTO:
|
|
"""Gets an image DTO."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_metadata(self, image_name: str) -> Optional[MetadataField]:
|
|
"""Gets an image's metadata."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_workflow(self, image_name: str) -> Optional[str]:
|
|
"""Gets an image's workflow."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_graph(self, image_name: str) -> Optional[str]:
|
|
"""Gets an image's workflow."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_path(self, image_name: str, thumbnail: bool = False) -> str:
|
|
"""Gets an image's path."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def validate_path(self, path: str) -> bool:
|
|
"""Validates an image's path."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_url(self, image_name: str, thumbnail: bool = False) -> str:
|
|
"""Gets an image's or thumbnail's URL."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_many(
|
|
self,
|
|
offset: int = 0,
|
|
limit: int = 10,
|
|
image_origin: Optional[ResourceOrigin] = None,
|
|
categories: Optional[list[ImageCategory]] = None,
|
|
is_intermediate: Optional[bool] = None,
|
|
board_id: Optional[str] = None,
|
|
) -> OffsetPaginatedResults[ImageDTO]:
|
|
"""Gets a paginated list of image DTOs."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete(self, image_name: str):
|
|
"""Deletes an image."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete_intermediates(self) -> int:
|
|
"""Deletes all intermediate images."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_intermediates_count(self) -> int:
|
|
"""Gets the number of intermediate images."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete_images_on_board(self, board_id: str):
|
|
"""Deletes all images on a board."""
|
|
pass
|