wehub-resource-sync
cddb07a176
docs / deploy (push) Has been cancelled
docs / changes (push) Has been cancelled
docs / check-and-build (push) Has been cancelled
build container image / cpu (push) Has been cancelled
build container image / cuda (push) Has been cancelled
build container image / rocm (push) Has been cancelled
frontend checks / frontend-checks (push) Has been cancelled
frontend tests / frontend-tests (push) Has been cancelled
lfs checks / lfs-check (push) Has been cancelled
python checks / python-checks (push) Has been cancelled
python tests / py3.12: macos-default (push) Has been cancelled
python tests / py3.11: windows-cpu (push) Has been cancelled
python tests / py3.12: windows-cpu (push) Has been cancelled
python tests / py3.11: linux-cpu (push) Has been cancelled
typegen checks / typegen-checks (push) Has been cancelled
uv lock checks / uv-lock-checks (push) Has been cancelled
openapi checks / openapi-checks (push) Has been cancelled
python tests / py3.11: macos-default (push) Has been cancelled
python tests / py3.12: linux-cpu (push) Has been cancelled
38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
# Copyright (c) 2022 Kyle Schouviller (https://github.com/kyle0654)
|
|
|
|
|
|
from invokeai.app.services.invocation_services import InvocationServices
|
|
|
|
|
|
class Invoker:
|
|
"""The invoker, used to execute invocations"""
|
|
|
|
services: InvocationServices
|
|
|
|
def __init__(self, services: InvocationServices):
|
|
self.services = services
|
|
self._start()
|
|
|
|
def __start_service(self, service) -> None:
|
|
# Call start() method on any services that have it
|
|
start_op = getattr(service, "start", None)
|
|
if callable(start_op):
|
|
start_op(self)
|
|
|
|
def __stop_service(self, service) -> None:
|
|
# Call stop() method on any services that have it
|
|
stop_op = getattr(service, "stop", None)
|
|
if callable(stop_op):
|
|
stop_op(self)
|
|
|
|
def _start(self) -> None:
|
|
"""Starts the invoker. This is called automatically when the invoker is created."""
|
|
for service in vars(self.services):
|
|
self.__start_service(getattr(self.services, service))
|
|
|
|
def stop(self) -> None:
|
|
"""Stops the invoker. A new invoker will have to be created to execute further."""
|
|
# First stop all services
|
|
for service in vars(self.services):
|
|
self.__stop_service(getattr(self.services, service))
|