File size: 2,103 Bytes
8a37e0a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from abc import ABC, abstractmethod
from typing import Optional, Union

from invokeai.app.invocations.baseinvocation import BaseInvocation, BaseInvocationOutput
from invokeai.app.services.invocation_cache.invocation_cache_common import InvocationCacheStatus


class InvocationCacheBase(ABC):
    """
    Base class for invocation caches.
    When an invocation is executed, it is hashed and its output stored in the cache.
    When new invocations are executed, if they are flagged with `use_cache`, they
    will attempt to pull their value from the cache before executing.

    Implementations should register for the `on_deleted` event of the `images` and `latents`
    services, and delete any cached outputs that reference the deleted image or latent.

    See the memory implementation for an example.

    Implementations should respect the `node_cache_size` configuration value, and skip all
    cache logic if the value is set to 0.
    """

    @abstractmethod
    def get(self, key: Union[int, str]) -> Optional[BaseInvocationOutput]:
        """Retrieves an invocation output from the cache"""
        pass

    @abstractmethod
    def save(self, key: Union[int, str], invocation_output: BaseInvocationOutput) -> None:
        """Stores an invocation output in the cache"""
        pass

    @abstractmethod
    def delete(self, key: Union[int, str]) -> None:
        """Deletes an invocation output from the cache"""
        pass

    @abstractmethod
    def clear(self) -> None:
        """Clears the cache"""
        pass

    @staticmethod
    @abstractmethod
    def create_key(invocation: BaseInvocation) -> int:
        """Gets the key for the invocation's cache item"""
        pass

    @abstractmethod
    def disable(self) -> None:
        """Disables the cache, overriding the max cache size"""
        pass

    @abstractmethod
    def enable(self) -> None:
        """Enables the cache, letting the the max cache size take effect"""
        pass

    @abstractmethod
    def get_status(self) -> InvocationCacheStatus:
        """Returns the status of the cache"""
        pass