Skip to content

pipeline

Actual host definition.

Will be launched as a subprocess.

ComfyUIHost

Bases: HostBase, IWorkfileHost, ILoadHost, IPublishHost

Main implementation of actual host operations.

Source code in client/ayon_comfyui/api/pipeline.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class ComfyUIHost(HostBase, IWorkfileHost, ILoadHost, IPublishHost):
    """Main implementation of actual host operations."""

    name = "comfyui"

    _last_path: str = ""

    def __init__(self):
        super().__init__()
        # Establish connection?
        # If not done here then deperecate this.

    def get_app_information(self) -> ApplicationInformation:
        """Return application information."""
        return ApplicationInformation(
            app_name="ComfyUI",
            app_version="N/A",  # TODO(@sas): request this in future
        )

    def get_workfile_extensions(self) -> list[str]:
        """Return list of workfile extensions."""
        return [".json"]

    def install(self) -> None:
        """Install host context."""

        register_loader_plugin_path(LOAD_PATH)
        register_creator_plugin_path(CREATE_PATH)

        # Pyblish
        pyblish.api.register_host("comfyui")
        pyblish.api.register_plugin_path(PUBLISH_PATH)

    def get_containers(self):
        return ls()

    def get_context_data(self):
        return self.stub.load_context()

    def update_context_data(self, data, changes):
        self.stub.imprint_context(data)

    def get_current_workfile(self):
        # Not too great, relies on a workfile having been opened
        return self._last_path or None

    def open_workfile(self, filepath):
        self.stub.load_workfile(filepath)
        return True

    def save_workfile(self, dst_path=None):
        workfile: str = self.stub.query_workfile()
        if workfile and isinstance(dst_path, str):
            with open(dst_path, mode="w", encoding="utf-8") as f:
                f.write(workfile)
                self.__class__._last_path = dst_path  # noqa: SLF001

    @property
    def stub(self) -> RPCStub:
        """Retrieve stub to interact with client."""
        return QRPCManager.get_instance().stub

stub property

Retrieve stub to interact with client.

get_app_information()

Return application information.

Source code in client/ayon_comfyui/api/pipeline.py 
56
57
58
59
60
61
def get_app_information(self) -> ApplicationInformation:
    """Return application information."""
    return ApplicationInformation(
        app_name="ComfyUI",
        app_version="N/A",  # TODO(@sas): request this in future
    )

get_workfile_extensions()

Return list of workfile extensions.

Source code in client/ayon_comfyui/api/pipeline.py 
63
64
65
def get_workfile_extensions(self) -> list[str]:
    """Return list of workfile extensions."""
    return [".json"]

install()

Install host context.

Source code in client/ayon_comfyui/api/pipeline.py 
67
68
69
70
71
72
73
74
75
def install(self) -> None:
    """Install host context."""

    register_loader_plugin_path(LOAD_PATH)
    register_creator_plugin_path(CREATE_PATH)

    # Pyblish
    pyblish.api.register_host("comfyui")
    pyblish.api.register_plugin_path(PUBLISH_PATH)

containerise(name, namespace, context, image_upload_info, loader=None, suffix='_CON')

Imprint layer with metadata.

Containerisation enables a tracking of version, author and origin for loaded assets.

Parameters:

Name Type Description Default
name str

Name of resulting assembly

 required
namespace str

Namespace under which to host container

 required
context dict

Asset information

 required
image_upload_info (dict, list[dict])

Uploaded image information, gotten back from /upload/image endpoint

 required
loader str

Name of loader used to produce this container.

 None
suffix str

Suffix of container, defaults to _CON.

 '_CON'

Returns:

Name Type Description
container str

Name of container assembly
Source code in client/ayon_comfyui/api/pipeline.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def containerise(  # noqa: PLR0913, PLR0917
    name: str,
    namespace: str,
    context: dict,
    image_upload_info: dict | list[dict],
    loader: str | None = None,
    suffix: str | None = "_CON",
) -> dict:
    """Imprint layer with metadata.

    Containerisation enables a tracking of version, author and origin
    for loaded assets.

    Arguments:
        name (str): Name of resulting assembly
        namespace (str): Namespace under which to host container
        context (dict): Asset information
        image_upload_info (dict, list[dict]): Uploaded image information,
                                  gotten back from /upload/image endpoint
        loader (str, optional): Name of loader used to produce this container.
        suffix (str, optional): Suffix of container, defaults to `_CON`.

    Returns:
        container (str): Name of container assembly
    """
    # TODO(@sas): Retrieve filename and store in container dict.

    container_name = name + suffix

    if isinstance(image_upload_info, dict):
        image_upload_info = [image_upload_info]

    data = {
        "schema": "ayon:container-3.0",
        "id": AYON_CONTAINER_ID,
        "name": name,
        "namespace": namespace,
        "loader": str(loader),
        "image_upload_info": image_upload_info,
        "representation": context["representation"]["id"],
        "container_uuid": str(uuid4()),
        "container_name": container_name,
    }
    stub = QRPCManager.get_instance().stub

    # TODO(@sas): Expand stub logic to keep track of containers separately,
    #             for my own sanity. That way, we can track representation
    #             instead of instance_id, and we might also want to enforce
    #             some uuid to keep track of every instance we make, since
    #             I can forsee people wanting to import the same image twice.
    #             It seems that regular implementations separate this through
    #             ls() and list_instances(), I guess. We will cleanly separate
    #             them.

    stub.add_containers(data)
    return data

list_instances()

Get cached instances in metadata.

Returns: List of dictionaries describing instances

Source code in client/ayon_comfyui/api/pipeline.py 
116
117
118
119
120
121
122
123
124
125
def list_instances() -> list[dict[str, Any]]:
    """Get cached instances in metadata.

    Returns:
    List of dictionaries describing instances
    """
    stub = QRPCManager.get_instance().stub
    instances = stub.list_instances()

    return instances or []

ls()

Yields valid containers.

Equivalent to ayon core api.ls().

Source code in client/ayon_comfyui/api/pipeline.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def ls() -> Generator[dict[str, str], None, None]:
    """Yields valid containers.

    Equivalent to ayon core api.ls().
    """
    try:
        rpcman = QRPCManager.get_instance()
        containers: list[dict[str, str]] = rpcman.stub.list_containers()
    except BaseException:  # noqa: BLE001
        return

    if not containers:
        return

    for container in containers:
        # Validate ID
        if not (con_id := container.get("id")) or "container" not in con_id:
            continue

        container["objectName"] = container.get("name")
        # Explicitly set namespace so that name shows up in UI
        container["namespace"] = (
            container.get("name")
            if container["namespace"] is None
            else container["namespace"]
        )
        yield container

uninstall()

Cleanup registration data.

Source code in client/ayon_comfyui/api/pipeline.py 
107
108
109
110
111
112
113
def uninstall() -> None:
    """Cleanup registration data."""
    pyblish.api.deregister_plugin_path(PUBLISH_PATH)
    pyblish.api.deregister_host("comfyui")

    deregister_loader_plugin_path(LOAD_PATH)
    deregister_creator_plugin_path(CREATE_PATH)