mirror of
https://github.com/invoke-ai/InvokeAI
synced 2024-08-30 20:32:17 +00:00
feat(nodes): organise/tidy
This commit is contained in:
psychedelicious
Kent Keirsey
parent
c0f132e41a
commit
52c9e6ec91
@ -1,42 +0,0 @@
|
||||
from fastapi import HTTPException, Path, Query
|
||||
from fastapi.routing import APIRouter
|
||||
from invokeai.app.services.results import ResultType, ResultWithSession
|
||||
from invokeai.app.services.item_storage import PaginatedResults
|
||||
|
||||
from ..dependencies import ApiDependencies
|
||||
|
||||
results_router = APIRouter(prefix="/v1/results", tags=["results"])
|
||||
|
||||
|
||||
@results_router.get("/{result_type}/{result_name}", operation_id="get_result")
|
||||
async def get_result(
|
||||
result_type: ResultType = Path(description="The type of result to get"),
|
||||
result_name: str = Path(description="The name of the result to get"),
|
||||
) -> ResultWithSession:
|
||||
"""Gets a result"""
|
||||
|
||||
result = ApiDependencies.invoker.services.results.get(
|
||||
result_id=result_name, result_type=result_type
|
||||
)
|
||||
|
||||
if result is not None:
|
||||
return result
|
||||
else:
|
||||
raise HTTPException(status_code=404)
|
||||
|
||||
|
||||
@results_router.get(
|
||||
"/",
|
||||
operation_id="list_results",
|
||||
responses={200: {"model": PaginatedResults[ResultWithSession]}},
|
||||
)
|
||||
async def list_results(
|
||||
result_type: ResultType = Query(description="The type of results to get"),
|
||||
page: int = Query(default=0, description="The page of results to get"),
|
||||
per_page: int = Query(default=10, description="The number of results per page"),
|
||||
) -> PaginatedResults[ResultWithSession]:
|
||||
"""Gets a list of results"""
|
||||
results = ApiDependencies.invoker.services.results.get_many(
|
||||
result_type=result_type, page=page, per_page=per_page
|
||||
)
|
||||
return results
|
||||
@ -1,6 +1,7 @@
|
||||
from abc import ABC, abstractmethod
|
||||
import datetime
|
||||
from typing import Optional
|
||||
from enum import Enum
|
||||
from typing import Optional, Type
|
||||
import sqlite3
|
||||
import threading
|
||||
from typing import Optional, Union
|
||||
@ -10,15 +11,49 @@ from invokeai.app.models.image import (
|
||||
ImageCategory,
|
||||
ImageType,
|
||||
)
|
||||
from invokeai.app.services.util.create_enum_table import create_enum_table
|
||||
from invokeai.app.services.models.image_record import (
|
||||
ImageRecord,
|
||||
deserialize_image_record,
|
||||
)
|
||||
|
||||
from invokeai.app.services.item_storage import PaginatedResults
|
||||
|
||||
|
||||
def create_sql_values_string_from_string_enum(enum: Type[Enum]):
|
||||
"""
|
||||
Creates a string of the form "('value1'), ('value2'), ..., ('valueN')" from a StrEnum.
|
||||
"""
|
||||
|
||||
delimiter = ", "
|
||||
values = [f"('{e.value}')" for e in enum]
|
||||
return delimiter.join(values)
|
||||
|
||||
|
||||
def create_enum_table(
|
||||
enum: Type[Enum],
|
||||
table_name: str,
|
||||
primary_key_name: str,
|
||||
cursor: sqlite3.Cursor,
|
||||
):
|
||||
"""
|
||||
Creates and populates a table to be used as a functional enum.
|
||||
"""
|
||||
|
||||
values_string = create_sql_values_string_from_string_enum(enum)
|
||||
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
CREATE TABLE IF NOT EXISTS {table_name} (
|
||||
{primary_key_name} TEXT PRIMARY KEY
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
INSERT OR IGNORE INTO {table_name} ({primary_key_name}) VALUES {values_string};
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
class ImageRecordStorageBase(ABC):
|
||||
"""Low-level service responsible for interfacing with the image record store."""
|
||||
|
||||
|
||||
@ -1,657 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum
|
||||
import enum
|
||||
import sqlite3
|
||||
import threading
|
||||
from typing import Optional, Type, TypeVar, Union
|
||||
from PIL.Image import Image as PILImage
|
||||
from pydantic import BaseModel, Field
|
||||
from torch import Tensor
|
||||
|
||||
from invokeai.app.services.item_storage import PaginatedResults
|
||||
|
||||
|
||||
"""
|
||||
Substantial proposed changes to the management of images and tensor.
|
||||
|
||||
tl;dr:
|
||||
With the upcoming move to latents-only nodes, we need to handle metadata differently. After struggling with this unsuccessfully - trying to smoosh it in to the existing setup - I believe we need to expand the scope of the refactor to include the management of images and latents - and make `latents` a special case of `tensor`.
|
||||
|
||||
full story:
|
||||
The consensus for tensor-only nodes' metadata was to traverse the execution graph and grab the core parameters to write to the image. This was straightforward, and I've written functions to find the nearest t2l/l2l, noise, and compel nodes and build the metadata from those.
|
||||
|
||||
But struggling to integrate this and the associated edge cases this brought up a number of issues deeper in the system (some of which I had previously implemented). The ImageStorageService is doing way too much, and we have a need to be able to retrieve sessions the session given image/latents id, which is not currently feasible due to SQLite's JSON parsing performance.
|
||||
|
||||
I made a new ResultsService and `results` table in the db to facilitate this. This first attempt failed because it doesn't handle uploads and leaves the codebase messy.
|
||||
|
||||
So I've spent the day trying to figure out to handle this in a sane way and think I've got something decent. I've described some changes to service bases and the database below.
|
||||
|
||||
The gist of it is to store the core parameters for an image in its metadata when the image is saved, but never to read from it. Instead, the same metadata is stored in the database, which will be set up for efficient access. So when a page of images is requested, the metadata comes from the db instead of a filesystem operation.
|
||||
|
||||
The URL generation responsibilities have been split off the image storage service in to a URL service. New database services/tables for images and tensor are added. These services will provide paginated images/tensors for the API to serve. This also paves the way for handling tensors as first-class outputs.
|
||||
"""
|
||||
|
||||
|
||||
# TODO: Make a new model for this
|
||||
class ResourceOrigin(str, Enum):
|
||||
"""The origin of a resource (eg image or tensor)."""
|
||||
|
||||
RESULTS = "results"
|
||||
UPLOADS = "uploads"
|
||||
INTERMEDIATES = "intermediates"
|
||||
|
||||
|
||||
class ImageKind(str, Enum):
|
||||
"""The kind of an image."""
|
||||
|
||||
IMAGE = "image"
|
||||
CONTROL_IMAGE = "control_image"
|
||||
|
||||
|
||||
class TensorKind(str, Enum):
|
||||
"""The kind of a tensor."""
|
||||
|
||||
IMAGE_TENSOR = "tensor"
|
||||
CONDITIONING = "conditioning"
|
||||
|
||||
|
||||
"""
|
||||
Core Generation Metadata Pydantic Model
|
||||
|
||||
I've already implemented the code to traverse a session to build this object.
|
||||
"""
|
||||
|
||||
|
||||
class CoreGenerationMetadata(BaseModel):
|
||||
"""Core generation metadata for an image/tensor generated in InvokeAI.
|
||||
|
||||
Generated by traversing the execution graph, collecting the parameters of the nearest ancestors of a given node.
|
||||
|
||||
Full metadata may be accessed by querying for the session in the `graph_executions` table.
|
||||
"""
|
||||
|
||||
positive_conditioning: Optional[str] = Field(
|
||||
description="The positive conditioning."
|
||||
)
|
||||
negative_conditioning: Optional[str] = Field(
|
||||
description="The negative conditioning."
|
||||
)
|
||||
width: Optional[int] = Field(description="Width of the image/tensor in pixels.")
|
||||
height: Optional[int] = Field(description="Height of the image/tensor in pixels.")
|
||||
seed: Optional[int] = Field(description="The seed used for noise generation.")
|
||||
cfg_scale: Optional[float] = Field(
|
||||
description="The classifier-free guidance scale."
|
||||
)
|
||||
steps: Optional[int] = Field(description="The number of steps used for inference.")
|
||||
scheduler: Optional[str] = Field(description="The scheduler used for inference.")
|
||||
model: Optional[str] = Field(description="The model used for inference.")
|
||||
strength: Optional[float] = Field(
|
||||
description="The strength used for image-to-image/tensor-to-tensor."
|
||||
)
|
||||
image: Optional[str] = Field(description="The ID of the initial image.")
|
||||
tensor: Optional[str] = Field(description="The ID of the initial tensor.")
|
||||
# Pending model refactor:
|
||||
# vae: Optional[str] = Field(description="The VAE used for decoding.")
|
||||
# unet: Optional[str] = Field(description="The UNet used dor inference.")
|
||||
# clip: Optional[str] = Field(description="The CLIP Encoder used for conditioning.")
|
||||
|
||||
|
||||
"""
|
||||
Minimal Uploads Metadata Model
|
||||
"""
|
||||
|
||||
|
||||
class UploadsMetadata(BaseModel):
|
||||
"""Limited metadata for an uploaded image/tensor."""
|
||||
|
||||
width: Optional[int] = Field(description="Width of the image/tensor in pixels.")
|
||||
height: Optional[int] = Field(description="Height of the image/tensor in pixels.")
|
||||
# The extra field will be the contents of the PNG file's tEXt chunk. It may have come
|
||||
# from another SD application or InvokeAI, so we need to make it very flexible. I think it's
|
||||
# best to just store it as a string and let the frontend parse it.
|
||||
# If the upload is a tensor type, this will be omitted.
|
||||
extra: Optional[str] = Field(
|
||||
description="Extra metadata, extracted from the PNG tEXt chunk."
|
||||
)
|
||||
|
||||
|
||||
"""
|
||||
Slimmed-down Image Storage Service Base
|
||||
- No longer lists images or generates URLs - only stores and retrieves images.
|
||||
- OSS implementation for disk storage
|
||||
"""
|
||||
|
||||
|
||||
class ImageStorageBase(ABC):
|
||||
"""Responsible for storing and retrieving images."""
|
||||
|
||||
@abstractmethod
|
||||
def save(
|
||||
self,
|
||||
image: PILImage,
|
||||
image_kind: ImageKind,
|
||||
origin: ResourceOrigin,
|
||||
context_id: str,
|
||||
node_id: str,
|
||||
metadata: CoreGenerationMetadata,
|
||||
) -> str:
|
||||
"""Saves an image and its thumbnail, returning its unique identifier."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get(self, id: str, thumbnail: bool = False) -> Union[PILImage, None]:
|
||||
"""Retrieves an image as a PIL Image."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, id: str) -> None:
|
||||
"""Deletes an image."""
|
||||
pass
|
||||
|
||||
|
||||
class TensorStorageBase(ABC):
|
||||
"""Responsible for storing and retrieving tensors."""
|
||||
|
||||
@abstractmethod
|
||||
def save(
|
||||
self,
|
||||
tensor: Tensor,
|
||||
tensor_kind: TensorKind,
|
||||
origin: ResourceOrigin,
|
||||
context_id: str,
|
||||
node_id: str,
|
||||
metadata: CoreGenerationMetadata,
|
||||
) -> str:
|
||||
"""Saves a tensor, returning its unique identifier."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get(self, id: str, thumbnail: bool = False) -> Union[Tensor, None]:
|
||||
"""Retrieves a tensor as a torch Tensor."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, id: str) -> None:
|
||||
"""Deletes a tensor."""
|
||||
pass
|
||||
|
||||
|
||||
"""
|
||||
New Url Service Base
|
||||
- Abstracts the logic for generating URLs out of the storage service
|
||||
- OSS implementation for locally-hosted URLs
|
||||
- Also provides a method to get the internal path to a resource (for OSS, the FS path)
|
||||
"""
|
||||
|
||||
|
||||
class ResourceLocationServiceBase(ABC):
|
||||
"""Responsible for locating resources (eg images or tensors)."""
|
||||
|
||||
@abstractmethod
|
||||
def get_url(self, id: str) -> str:
|
||||
"""Gets the URL for a resource."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_path(self, id: str) -> str:
|
||||
"""Gets the path for a resource."""
|
||||
pass
|
||||
|
||||
|
||||
"""
|
||||
New Images Database Service Base
|
||||
|
||||
This is a new service that will be responsible for the new `images` table(s):
|
||||
- Storing images in the table
|
||||
- Retrieving individual images and pages of images
|
||||
- Deleting individual images
|
||||
|
||||
Operations will typically use joins with the various `images` tables.
|
||||
"""
|
||||
|
||||
|
||||
class ImagesDbServiceBase(ABC):
|
||||
"""Responsible for interfacing with `images` table."""
|
||||
|
||||
class GeneratedImageEntity(BaseModel):
|
||||
id: str = Field(description="The unique identifier for the image.")
|
||||
session_id: str = Field(description="The session ID.")
|
||||
node_id: str = Field(description="The node ID.")
|
||||
metadata: CoreGenerationMetadata = Field(
|
||||
description="The metadata for the image."
|
||||
)
|
||||
|
||||
class UploadedImageEntity(BaseModel):
|
||||
id: str = Field(description="The unique identifier for the image.")
|
||||
metadata: UploadsMetadata = Field(description="The metadata for the image.")
|
||||
|
||||
@abstractmethod
|
||||
def get(self, id: str) -> Union[GeneratedImageEntity, UploadedImageEntity, None]:
|
||||
"""Gets an image from the `images` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_many(
|
||||
self, image_kind: ImageKind, page: int = 0, per_page: int = 10
|
||||
) -> PaginatedResults[Union[GeneratedImageEntity, UploadedImageEntity]]:
|
||||
"""Gets a page of images from the `images` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, id: str) -> None:
|
||||
"""Deletes an image from the `images` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def set(
|
||||
self,
|
||||
id: str,
|
||||
image_kind: ImageKind,
|
||||
session_id: Optional[str],
|
||||
node_id: Optional[str],
|
||||
metadata: CoreGenerationMetadata | UploadsMetadata,
|
||||
) -> None:
|
||||
"""Sets an image in the `images` table."""
|
||||
pass
|
||||
|
||||
|
||||
"""
|
||||
New Tensor Database Service Base
|
||||
|
||||
This is a new service that will be responsible for the new `tensor` table:
|
||||
- Storing tensor in the table
|
||||
- Retrieving individual tensor and pages of tensor
|
||||
- Deleting individual tensor
|
||||
|
||||
Operations will always use joins with the `tensor_metadata` table.
|
||||
"""
|
||||
|
||||
|
||||
class TensorDbServiceBase(ABC):
|
||||
"""Responsible for interfacing with `tensor` table."""
|
||||
|
||||
class GeneratedTensorEntity(BaseModel):
|
||||
id: str = Field(description="The unique identifier for the tensor.")
|
||||
session_id: str = Field(description="The session ID.")
|
||||
node_id: str = Field(description="The node ID.")
|
||||
metadata: CoreGenerationMetadata = Field(
|
||||
description="The metadata for the tensor."
|
||||
)
|
||||
|
||||
class UploadedTensorEntity(BaseModel):
|
||||
id: str = Field(description="The unique identifier for the tensor.")
|
||||
metadata: UploadsMetadata = Field(description="The metadata for the tensor.")
|
||||
|
||||
@abstractmethod
|
||||
def get(self, id: str) -> Union[GeneratedTensorEntity, UploadedTensorEntity, None]:
|
||||
"""Gets a tensor from the `tensor` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_many(
|
||||
self, tensor_kind: TensorKind, page: int = 0, per_page: int = 10
|
||||
) -> PaginatedResults[Union[GeneratedTensorEntity, UploadedTensorEntity]]:
|
||||
"""Gets a page of tensor from the `tensor` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, id: str) -> None:
|
||||
"""Deletes a tensor from the `tensor` table."""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def set(
|
||||
self,
|
||||
id: str,
|
||||
tensor_kind: TensorKind,
|
||||
session_id: Optional[str],
|
||||
node_id: Optional[str],
|
||||
metadata: CoreGenerationMetadata | UploadsMetadata,
|
||||
) -> None:
|
||||
"""Sets a tensor in the `tensor` table."""
|
||||
pass
|
||||
|
||||
|
||||
"""
|
||||
Database Changes
|
||||
|
||||
The existing tables will remain as-is, new tables will be added.
|
||||
|
||||
Tensor now also have the same types as images - `results`, `intermediates`, `uploads`. Storage, retrieval, and operations may diverge from images in the future, so they are managed separately.
|
||||
|
||||
A few `images` tables are created to store all images:
|
||||
- `results` and `intermediates` images have additional data: `session_id` and `node_id`, and may be further differentiated in the future. For this reason, they each get their own table.
|
||||
- `uploads` do not get their own table, as they are never going to have more than an `id`, `image_kind` and `timestamp`.
|
||||
- `images_metadata` holds the same image metadata that is written to the image. This table, along with the URL service, allow us to more efficiently serve images without having to read the image from storage.
|
||||
|
||||
The same tables are made for `tensor` and for the moment, implementation is expected to be identical.
|
||||
|
||||
Schemas for each table below.
|
||||
|
||||
Insertions and updates of ancillary tables (e.g. `results_images`, `images_metadata`, etc) will need to be done manually in the services, but should be straightforward. Deletion via cascading will be handled by the database.
|
||||
"""
|
||||
|
||||
|
||||
def create_sql_values_string_from_string_enum(enum: Type[Enum]):
|
||||
"""
|
||||
Creates a string of the form "('value1'), ('value2'), ..., ('valueN')" from a StrEnum.
|
||||
"""
|
||||
|
||||
delimiter = ", "
|
||||
values = [f"('{e.value}')" for e in enum]
|
||||
return delimiter.join(values)
|
||||
|
||||
|
||||
def create_sql_table_from_enum(
|
||||
enum: Type[Enum],
|
||||
table_name: str,
|
||||
primary_key_name: str,
|
||||
cursor: sqlite3.Cursor,
|
||||
lock: threading.Lock,
|
||||
):
|
||||
"""
|
||||
Creates and populates a table to be used as a functional enum.
|
||||
"""
|
||||
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
values_string = create_sql_values_string_from_string_enum(enum)
|
||||
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
CREATE TABLE IF NOT EXISTS {table_name} (
|
||||
{primary_key_name} TEXT PRIMARY KEY
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
INSERT OR IGNORE INTO {table_name} ({primary_key_name}) VALUES {values_string};
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
"""
|
||||
`resource_origins` functions as an enum for the ResourceOrigin model.
|
||||
"""
|
||||
|
||||
|
||||
def create_resource_origins_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
create_sql_table_from_enum(
|
||||
enum=ResourceOrigin,
|
||||
table_name="resource_origins",
|
||||
primary_key_name="origin_name",
|
||||
cursor=cursor,
|
||||
lock=lock,
|
||||
)
|
||||
|
||||
|
||||
"""
|
||||
`image_kinds` functions as an enum for the ImageType model.
|
||||
"""
|
||||
|
||||
|
||||
def create_image_kinds_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
create_sql_table_from_enum(
|
||||
enum=ImageKind,
|
||||
table_name="image_kinds",
|
||||
primary_key_name="kind_name",
|
||||
cursor=cursor,
|
||||
lock=lock,
|
||||
)
|
||||
|
||||
|
||||
"""
|
||||
`tensor_kinds` functions as an enum for the TensorType model.
|
||||
"""
|
||||
|
||||
|
||||
def create_tensor_kinds_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
create_sql_table_from_enum(
|
||||
enum=TensorKind,
|
||||
table_name="tensor_kinds",
|
||||
primary_key_name="kind_name",
|
||||
cursor=cursor,
|
||||
lock=lock,
|
||||
)
|
||||
|
||||
|
||||
"""
|
||||
`images` stores all images, regardless of type
|
||||
"""
|
||||
|
||||
|
||||
def create_images_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS images (
|
||||
id TEXT PRIMARY KEY,
|
||||
origin TEXT,
|
||||
image_kind TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY(origin) REFERENCES resource_origins(origin_name),
|
||||
FOREIGN KEY(image_kind) REFERENCES image_kinds(kind_name)
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_images_id ON images(id);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE INDEX IF NOT EXISTS idx_images_origin ON images(origin);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE INDEX IF NOT EXISTS idx_images_image_kind ON images(image_kind);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
"""
|
||||
`image_results` stores additional data specific to `results` images.
|
||||
"""
|
||||
|
||||
|
||||
def create_image_results_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS image_results (
|
||||
images_id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
FOREIGN KEY(images_id) REFERENCES images(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_image_results_images_id ON image_results(id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
"""
|
||||
`image_intermediates` stores additional data specific to `intermediates` images
|
||||
"""
|
||||
|
||||
|
||||
def create_image_intermediates_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS image_intermediates (
|
||||
images_id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
FOREIGN KEY(images_id) REFERENCES images(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_image_intermediates_images_id ON image_intermediates(id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
"""
|
||||
`images_metadata` stores basic metadata for any image type
|
||||
"""
|
||||
|
||||
|
||||
def create_images_metadata_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS images_metadata (
|
||||
images_id TEXT PRIMARY KEY,
|
||||
metadata TEXT,
|
||||
FOREIGN KEY(images_id) REFERENCES images(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_images_metadata_images_id ON images_metadata(images_id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
# `tensor` table: stores references to tensor
|
||||
|
||||
|
||||
def create_tensors_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS tensors (
|
||||
id TEXT PRIMARY KEY,
|
||||
origin TEXT,
|
||||
tensor_kind TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY(origin) REFERENCES resource_origins(origin_name),
|
||||
FOREIGN KEY(tensor_kind) REFERENCES tensor_kinds(kind_name),
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_tensors_id ON tensors(id);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE INDEX IF NOT EXISTS idx_tensors_origin ON tensors(origin);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE INDEX IF NOT EXISTS idx_tensors_tensor_kind ON tensors(tensor_kind);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
# `results_tensor` stores additional data specific to `result` tensor
|
||||
|
||||
|
||||
def create_tensor_results_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS tensor_results (
|
||||
tensor_id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
FOREIGN KEY(tensor_id) REFERENCES tensors(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_tensor_results_tensor_id ON tensor_results(tensor_id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
# `tensor_intermediates` stores additional data specific to `intermediate` tensor
|
||||
|
||||
|
||||
def create_tensor_intermediates_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS tensor_intermediates (
|
||||
tensor_id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
FOREIGN KEY(tensor_id) REFERENCES tensors(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_tensor_intermediates_tensor_id ON tensor_intermediates(tensor_id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
# `tensors_metadata` table: stores generated/transformed metadata for tensor
|
||||
|
||||
|
||||
def create_tensors_metadata_table(cursor: sqlite3.Cursor, lock: threading.Lock):
|
||||
try:
|
||||
lock.acquire()
|
||||
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS tensors_metadata (
|
||||
tensor_id TEXT PRIMARY KEY,
|
||||
metadata TEXT,
|
||||
FOREIGN KEY(tensor_id) REFERENCES tensors(id) ON DELETE CASCADE
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_tensors_metadata_tensor_id ON tensors_metadata(tensor_id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
lock.release()
|
||||
@ -1,466 +0,0 @@
|
||||
from enum import Enum
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
import json
|
||||
import sqlite3
|
||||
from threading import Lock
|
||||
from typing import Any, Union
|
||||
|
||||
import networkx as nx
|
||||
|
||||
from pydantic import BaseModel, Field, parse_obj_as, parse_raw_as
|
||||
from invokeai.app.invocations.image import ImageOutput
|
||||
from invokeai.app.services.graph import Edge, GraphExecutionState
|
||||
from invokeai.app.invocations.latent import LatentsOutput
|
||||
from invokeai.app.services.item_storage import PaginatedResults
|
||||
from invokeai.app.util.misc import get_timestamp
|
||||
|
||||
|
||||
class ResultType(str, Enum):
|
||||
image_output = "image_output"
|
||||
latents_output = "latents_output"
|
||||
|
||||
|
||||
class Result(BaseModel):
|
||||
"""A session result"""
|
||||
|
||||
id: str = Field(description="Result ID")
|
||||
session_id: str = Field(description="Session ID")
|
||||
node_id: str = Field(description="Node ID")
|
||||
data: Union[LatentsOutput, ImageOutput] = Field(description="The result data")
|
||||
|
||||
|
||||
class ResultWithSession(BaseModel):
|
||||
"""A result with its session"""
|
||||
|
||||
result: Result = Field(description="The result")
|
||||
session: GraphExecutionState = Field(description="The session")
|
||||
|
||||
|
||||
# Create a directed graph
|
||||
from typing import Any, TypedDict, Union
|
||||
from networkx import DiGraph
|
||||
import networkx as nx
|
||||
import json
|
||||
|
||||
|
||||
# We need to use a loose class for nodes to allow for graceful parsing - we cannot use the stricter
|
||||
# model used by the system, because we may be a graph in an old format. We can, however, use the
|
||||
# Edge model, because the edge format does not change.
|
||||
class LooseGraph(BaseModel):
|
||||
id: str
|
||||
nodes: dict[str, dict[str, Any]]
|
||||
edges: list[Edge]
|
||||
|
||||
|
||||
# An intermediate type used during parsing
|
||||
class NearestAncestor(TypedDict):
|
||||
node_id: str
|
||||
metadata: dict[str, Any]
|
||||
|
||||
|
||||
# The ancestor types that contain the core metadata
|
||||
ANCESTOR_TYPES = ['t2l', 'l2l']
|
||||
|
||||
# The core metadata parameters in the ancestor types
|
||||
ANCESTOR_PARAMS = ['steps', 'model', 'cfg_scale', 'scheduler', 'strength']
|
||||
|
||||
# The core metadata parameters in the noise node
|
||||
NOISE_FIELDS = ['seed', 'width', 'height']
|
||||
|
||||
# Find nearest t2l or l2l ancestor from a given l2i node
|
||||
def find_nearest_ancestor(G: DiGraph, node_id: str) -> Union[NearestAncestor, None]:
|
||||
"""Returns metadata for the nearest ancestor of a given node.
|
||||
|
||||
Parameters:
|
||||
G (DiGraph): A directed graph.
|
||||
node_id (str): The ID of the starting node.
|
||||
|
||||
Returns:
|
||||
NearestAncestor | None: An object with the ID and metadata of the nearest ancestor.
|
||||
"""
|
||||
|
||||
# Retrieve the node from the graph
|
||||
node = G.nodes[node_id]
|
||||
|
||||
# If the node type is one of the core metadata node types, gather necessary metadata and return
|
||||
if node.get('type') in ANCESTOR_TYPES:
|
||||
parsed_metadata = {param: val for param, val in node.items() if param in ANCESTOR_PARAMS}
|
||||
return NearestAncestor(node_id=node_id, metadata=parsed_metadata)
|
||||
|
||||
|
||||
# Else, look for the ancestor in the predecessor nodes
|
||||
for predecessor in G.predecessors(node_id):
|
||||
result = find_nearest_ancestor(G, predecessor)
|
||||
if result:
|
||||
return result
|
||||
|
||||
# If there are no valid ancestors, return None
|
||||
return None
|
||||
|
||||
|
||||
def get_additional_metadata(graph: LooseGraph, node_id: str) -> Union[dict[str, Any], None]:
|
||||
"""Collects additional metadata from nodes connected to a given node.
|
||||
|
||||
Parameters:
|
||||
graph (LooseGraph): The graph.
|
||||
node_id (str): The ID of the node.
|
||||
|
||||
Returns:
|
||||
dict | None: A dictionary containing additional metadata.
|
||||
"""
|
||||
|
||||
metadata = {}
|
||||
|
||||
# Iterate over all edges in the graph
|
||||
for edge in graph.edges:
|
||||
dest_node_id = edge.destination.node_id
|
||||
dest_field = edge.destination.field
|
||||
source_node = graph.nodes[edge.source.node_id]
|
||||
|
||||
# If the destination node ID matches the given node ID, gather necessary metadata
|
||||
if dest_node_id == node_id:
|
||||
# If the destination field is 'positive_conditioning', add the 'prompt' from the source node
|
||||
if dest_field == 'positive_conditioning':
|
||||
metadata['positive_conditioning'] = source_node.get('prompt')
|
||||
# If the destination field is 'negative_conditioning', add the 'prompt' from the source node
|
||||
if dest_field == 'negative_conditioning':
|
||||
metadata['negative_conditioning'] = source_node.get('prompt')
|
||||
# If the destination field is 'noise', add the core noise fields from the source node
|
||||
if dest_field == 'noise':
|
||||
for field in NOISE_FIELDS:
|
||||
metadata[field] = source_node.get(field)
|
||||
return metadata
|
||||
|
||||
def build_core_metadata(graph_raw: str, node_id: str) -> Union[dict, None]:
|
||||
"""Builds the core metadata for a given node.
|
||||
|
||||
Parameters:
|
||||
graph_raw (str): The graph structure as a raw string.
|
||||
node_id (str): The ID of the node.
|
||||
|
||||
Returns:
|
||||
dict | None: A dictionary containing core metadata.
|
||||
"""
|
||||
|
||||
# Create a directed graph to facilitate traversal
|
||||
G = nx.DiGraph()
|
||||
|
||||
# Convert the raw graph string into a JSON object
|
||||
graph = parse_obj_as(LooseGraph, graph_raw)
|
||||
|
||||
# Add nodes and edges to the graph
|
||||
for node_id, node_data in graph.nodes.items():
|
||||
G.add_node(node_id, **node_data)
|
||||
for edge in graph.edges:
|
||||
G.add_edge(edge.source.node_id, edge.destination.node_id)
|
||||
|
||||
# Find the nearest ancestor of the given node
|
||||
ancestor = find_nearest_ancestor(G, node_id)
|
||||
|
||||
# If no ancestor was found, return None
|
||||
if ancestor is None:
|
||||
return None
|
||||
|
||||
metadata = ancestor['metadata']
|
||||
ancestor_id = ancestor['node_id']
|
||||
|
||||
# Get additional metadata related to the ancestor
|
||||
addl_metadata = get_additional_metadata(graph, ancestor_id)
|
||||
|
||||
# If additional metadata was found, add it to the main metadata
|
||||
if addl_metadata is not None:
|
||||
metadata.update(addl_metadata)
|
||||
|
||||
return metadata
|
||||
|
||||
|
||||
|
||||
class ResultsServiceABC(ABC):
|
||||
"""The Results service is responsible for retrieving results."""
|
||||
|
||||
@abstractmethod
|
||||
def get(
|
||||
self, result_id: str, result_type: ResultType
|
||||
) -> Union[ResultWithSession, None]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_many(
|
||||
self, result_type: ResultType, page: int = 0, per_page: int = 10
|
||||
) -> PaginatedResults[ResultWithSession]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def search(
|
||||
self, query: str, page: int = 0, per_page: int = 10
|
||||
) -> PaginatedResults[ResultWithSession]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def handle_graph_execution_state_change(self, session: GraphExecutionState) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class SqliteResultsService(ResultsServiceABC):
|
||||
"""SQLite implementation of the Results service."""
|
||||
|
||||
_filename: str
|
||||
_conn: sqlite3.Connection
|
||||
_cursor: sqlite3.Cursor
|
||||
_lock: Lock
|
||||
|
||||
def __init__(self, filename: str):
|
||||
super().__init__()
|
||||
|
||||
self._filename = filename
|
||||
self._lock = Lock()
|
||||
|
||||
self._conn = sqlite3.connect(
|
||||
self._filename, check_same_thread=False
|
||||
) # TODO: figure out a better threading solution
|
||||
self._cursor = self._conn.cursor()
|
||||
|
||||
self._create_table()
|
||||
|
||||
def _create_table(self):
|
||||
try:
|
||||
self._lock.acquire()
|
||||
self._cursor.execute(
|
||||
"""--sql
|
||||
CREATE TABLE IF NOT EXISTS results (
|
||||
id TEXT PRIMARY KEY, -- the result's name
|
||||
result_type TEXT, -- `image_output` | `latents_output`
|
||||
node_id TEXT, -- the node that produced this result
|
||||
session_id TEXT, -- the session that produced this result
|
||||
created_at INTEGER, -- the time at which this result was created
|
||||
data TEXT -- the result itself
|
||||
);
|
||||
"""
|
||||
)
|
||||
self._cursor.execute(
|
||||
"""--sql
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_result_id ON results(id);
|
||||
"""
|
||||
)
|
||||
finally:
|
||||
self._lock.release()
|
||||
|
||||
def _parse_joined_result(self, result_row: Any, column_names: list[str]):
|
||||
result_raw = {}
|
||||
session_raw = {}
|
||||
|
||||
for idx, name in enumerate(column_names):
|
||||
if name == "session":
|
||||
session_raw = json.loads(result_row[idx])
|
||||
elif name == "data":
|
||||
result_raw[name] = json.loads(result_row[idx])
|
||||
else:
|
||||
result_raw[name] = result_row[idx]
|
||||
|
||||
graph_raw = session_raw['execution_graph']
|
||||
|
||||
result = parse_obj_as(Result, result_raw)
|
||||
session = parse_obj_as(GraphExecutionState, session_raw)
|
||||
|
||||
m = build_core_metadata(graph_raw, result.node_id)
|
||||
print(m)
|
||||
|
||||
# g = session.execution_graph.nx_graph()
|
||||
# ancestors = nx.dag.ancestors(g, result.node_id)
|
||||
|
||||
# nodes = [session.execution_graph.get_node(result.node_id)]
|
||||
# for ancestor in ancestors:
|
||||
# nodes.append(session.execution_graph.get_node(ancestor))
|
||||
|
||||
# filtered_nodes = filter(lambda n: n.type in NODE_TYPE_ALLOWLIST, nodes)
|
||||
# print(list(map(lambda n: n.dict(), filtered_nodes)))
|
||||
# metadata = {}
|
||||
# for node in nodes:
|
||||
# if (node.type in ['txt2img', 'img2img',])
|
||||
# for field, value in node.dict().items():
|
||||
# if field not in ['type', 'id']:
|
||||
# if field not in metadata:
|
||||
# metadata[field] = value
|
||||
|
||||
# print(ancestors)
|
||||
# print(nodes)
|
||||
# print(metadata)
|
||||
|
||||
# for node in nodes:
|
||||
# print(node.dict())
|
||||
|
||||
# print(nodes)
|
||||
|
||||
return ResultWithSession(
|
||||
result=result,
|
||||
session=session,
|
||||
)
|
||||
|
||||
def get(
|
||||
self, result_id: str, result_type: ResultType
|
||||
) -> Union[ResultWithSession, None]:
|
||||
"""Retrieves a result by ID and type."""
|
||||
try:
|
||||
self._lock.acquire()
|
||||
self._cursor.execute(
|
||||
"""--sql
|
||||
SELECT
|
||||
results.id AS id,
|
||||
results.result_type AS result_type,
|
||||
results.node_id AS node_id,
|
||||
results.session_id AS session_id,
|
||||
results.data AS data,
|
||||
graph_executions.item AS session
|
||||
FROM results
|
||||
JOIN graph_executions ON results.session_id = graph_executions.id
|
||||
WHERE results.id = ? AND results.result_type = ?
|
||||
""",
|
||||
(result_id, result_type),
|
||||
)
|
||||
|
||||
result_row = self._cursor.fetchone()
|
||||
|
||||
if result_row is None:
|
||||
return None
|
||||
|
||||
column_names = list(map(lambda x: x[0], self._cursor.description))
|
||||
result_parsed = self._parse_joined_result(result_row, column_names)
|
||||
finally:
|
||||
self._lock.release()
|
||||
|
||||
if not result_parsed:
|
||||
return None
|
||||
|
||||
return result_parsed
|
||||
|
||||
def get_many(
|
||||
self,
|
||||
result_type: ResultType,
|
||||
page: int = 0,
|
||||
per_page: int = 10,
|
||||
) -> PaginatedResults[ResultWithSession]:
|
||||
"""Lists results of a given type."""
|
||||
try:
|
||||
self._lock.acquire()
|
||||
|
||||
self._cursor.execute(
|
||||
f"""--sql
|
||||
SELECT
|
||||
results.id AS id,
|
||||
results.result_type AS result_type,
|
||||
results.node_id AS node_id,
|
||||
results.session_id AS session_id,
|
||||
results.data AS data,
|
||||
graph_executions.item AS session
|
||||
FROM results
|
||||
JOIN graph_executions ON results.session_id = graph_executions.id
|
||||
WHERE results.result_type = ?
|
||||
LIMIT ? OFFSET ?;
|
||||
""",
|
||||
(result_type.value, per_page, page * per_page),
|
||||
)
|
||||
|
||||
result_rows = self._cursor.fetchall()
|
||||
column_names = list(map(lambda c: c[0], self._cursor.description))
|
||||
|
||||
result_parsed = []
|
||||
|
||||
for result_row in result_rows:
|
||||
result_parsed.append(
|
||||
self._parse_joined_result(result_row, column_names)
|
||||
)
|
||||
|
||||
self._cursor.execute("""SELECT count(*) FROM results;""")
|
||||
count = self._cursor.fetchone()[0]
|
||||
finally:
|
||||
self._lock.release()
|
||||
|
||||
pageCount = int(count / per_page) + 1
|
||||
|
||||
return PaginatedResults[ResultWithSession](
|
||||
items=result_parsed,
|
||||
page=page,
|
||||
pages=pageCount,
|
||||
per_page=per_page,
|
||||
total=count,
|
||||
)
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
page: int = 0,
|
||||
per_page: int = 10,
|
||||
) -> PaginatedResults[ResultWithSession]:
|
||||
"""Finds results by query."""
|
||||
try:
|
||||
self._lock.acquire()
|
||||
self._cursor.execute(
|
||||
"""--sql
|
||||
SELECT results.data, graph_executions.item
|
||||
FROM results
|
||||
JOIN graph_executions ON results.session_id = graph_executions.id
|
||||
WHERE item LIKE ?
|
||||
LIMIT ? OFFSET ?;
|
||||
""",
|
||||
(f"%{query}%", per_page, page * per_page),
|
||||
)
|
||||
|
||||
result_rows = self._cursor.fetchall()
|
||||
|
||||
items = list(
|
||||
map(
|
||||
lambda r: ResultWithSession(
|
||||
result=parse_raw_as(Result, r[0]),
|
||||
session=parse_raw_as(GraphExecutionState, r[1]),
|
||||
),
|
||||
result_rows,
|
||||
)
|
||||
)
|
||||
self._cursor.execute(
|
||||
"""--sql
|
||||
SELECT count(*) FROM results WHERE item LIKE ?;
|
||||
""",
|
||||
(f"%{query}%",),
|
||||
)
|
||||
count = self._cursor.fetchone()[0]
|
||||
finally:
|
||||
self._lock.release()
|
||||
|
||||
pageCount = int(count / per_page) + 1
|
||||
|
||||
return PaginatedResults[ResultWithSession](
|
||||
items=items, page=page, pages=pageCount, per_page=per_page, total=count
|
||||
)
|
||||
|
||||
def handle_graph_execution_state_change(self, session: GraphExecutionState) -> None:
|
||||
"""Updates the results table with the results from the session."""
|
||||
with self._conn as conn:
|
||||
for node_id, result in session.results.items():
|
||||
# We'll only process 'image_output' or 'latents_output'
|
||||
if result.type not in ["image_output", "latents_output"]:
|
||||
continue
|
||||
|
||||
# The id depends on the result type
|
||||
if result.type == "image_output":
|
||||
id = result.image.image_name
|
||||
result_type = "image_output"
|
||||
else:
|
||||
id = result.latents.latents_name
|
||||
result_type = "latents_output"
|
||||
|
||||
# Insert the result into the results table, ignoring if it already exists
|
||||
conn.execute(
|
||||
"""--sql
|
||||
INSERT OR IGNORE INTO results (id, result_type, node_id, session_id, created_at, data)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
id,
|
||||
result_type,
|
||||
node_id,
|
||||
session.id,
|
||||
get_timestamp(),
|
||||
result.json(),
|
||||
),
|
||||
)
|
||||
@ -1,39 +0,0 @@
|
||||
from enum import Enum
|
||||
import sqlite3
|
||||
from typing import Type
|
||||
|
||||
|
||||
def create_sql_values_string_from_string_enum(enum: Type[Enum]):
|
||||
"""
|
||||
Creates a string of the form "('value1'), ('value2'), ..., ('valueN')" from a StrEnum.
|
||||
"""
|
||||
|
||||
delimiter = ", "
|
||||
values = [f"('{e.value}')" for e in enum]
|
||||
return delimiter.join(values)
|
||||
|
||||
|
||||
def create_enum_table(
|
||||
enum: Type[Enum],
|
||||
table_name: str,
|
||||
primary_key_name: str,
|
||||
cursor: sqlite3.Cursor,
|
||||
):
|
||||
"""
|
||||
Creates and populates a table to be used as a functional enum.
|
||||
"""
|
||||
|
||||
values_string = create_sql_values_string_from_string_enum(enum)
|
||||
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
CREATE TABLE IF NOT EXISTS {table_name} (
|
||||
{primary_key_name} TEXT PRIMARY KEY
|
||||
);
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
f"""--sql
|
||||
INSERT OR IGNORE INTO {table_name} ({primary_key_name}) VALUES {values_string};
|
||||
"""
|
||||
)
|
||||
Loading…
Reference in New Issue
Block a user