Spaces:
Runtime error
Runtime error
File size: 5,823 Bytes
6831a54 |
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 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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 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 |
from __future__ import annotations
import os
import time
from modules import timer
from modules import initialize_util
from modules import initialize
from threading import Thread
from modules_forge.initialization import initialize_forge
from modules_forge import main_thread
startup_timer = timer.startup_timer
startup_timer.record("launcher")
initialize_forge()
initialize.imports()
initialize.check_versions()
initialize.initialize()
def create_api(app):
from modules.api.api import Api
from modules.call_queue import queue_lock
api = Api(app, queue_lock)
return api
def api_only_worker():
from fastapi import FastAPI
from modules.shared_cmd_options import cmd_opts
app = FastAPI()
initialize_util.setup_middleware(app)
api = create_api(app)
from modules import script_callbacks
script_callbacks.before_ui_callback()
script_callbacks.app_started_callback(None, app)
print(f"Startup time: {startup_timer.summary()}.")
api.launch(
server_name=initialize_util.gradio_server_name(),
port=cmd_opts.port if cmd_opts.port else 7861,
root_path=f"/{cmd_opts.subpath}" if cmd_opts.subpath else ""
)
def webui_worker():
from modules.shared_cmd_options import cmd_opts
launch_api = cmd_opts.api
from modules import shared, ui_tempdir, script_callbacks, ui, progress, ui_extra_networks
while 1:
if shared.opts.clean_temp_dir_at_start:
ui_tempdir.cleanup_tmpdr()
startup_timer.record("cleanup temp dir")
script_callbacks.before_ui_callback()
startup_timer.record("scripts before_ui_callback")
shared.demo = ui.create_ui()
startup_timer.record("create ui")
if not cmd_opts.no_gradio_queue:
shared.demo.queue(64)
gradio_auth_creds = list(initialize_util.get_gradio_auth_creds()) or None
auto_launch_browser = False
if os.getenv('SD_WEBUI_RESTARTING') != '1':
if shared.opts.auto_launch_browser == "Remote" or cmd_opts.autolaunch:
auto_launch_browser = True
elif shared.opts.auto_launch_browser == "Local":
auto_launch_browser = not cmd_opts.webui_is_non_local
from modules_forge.forge_canvas.canvas import canvas_js_root_path
app, local_url, share_url = shared.demo.launch(
share=cmd_opts.share,
server_name=initialize_util.gradio_server_name(),
server_port=cmd_opts.port,
ssl_keyfile=cmd_opts.tls_keyfile,
ssl_certfile=cmd_opts.tls_certfile,
ssl_verify=cmd_opts.disable_tls_verify,
debug=cmd_opts.gradio_debug,
auth=gradio_auth_creds,
inbrowser=auto_launch_browser,
prevent_thread_lock=True,
allowed_paths=cmd_opts.gradio_allowed_path + [canvas_js_root_path],
app_kwargs={
"docs_url": "/docs",
"redoc_url": "/redoc",
},
root_path=f"/{cmd_opts.subpath}" if cmd_opts.subpath else "",
)
startup_timer.record("gradio launch")
# gradio uses a very open CORS policy via app.user_middleware, which makes it possible for
# an attacker to trick the user into opening a malicious HTML page, which makes a request to the
# running web ui and do whatever the attacker wants, including installing an extension and
# running its code. We disable this here. Suggested by RyotaK.
app.user_middleware = [x for x in app.user_middleware if x.cls.__name__ != 'CORSMiddleware']
initialize_util.setup_middleware(app)
progress.setup_progress_api(app)
ui.setup_ui_api(app)
if launch_api:
create_api(app)
ui_extra_networks.add_pages_to_demo(app)
startup_timer.record("add APIs")
with startup_timer.subcategory("app_started_callback"):
script_callbacks.app_started_callback(shared.demo, app)
timer.startup_record = startup_timer.dump()
print(f"Startup time: {startup_timer.summary()}.")
try:
while True:
server_command = shared.state.wait_for_server_command(timeout=5)
if server_command:
if server_command in ("stop", "restart"):
break
else:
print(f"Unknown server command: {server_command}")
except KeyboardInterrupt:
print('Caught KeyboardInterrupt, stopping...')
server_command = "stop"
if server_command == "stop":
print("Stopping server...")
# If we catch a keyboard interrupt, we want to stop the server and exit.
shared.demo.close()
break
# disable auto launch webui in browser for subsequent UI Reload
os.environ.setdefault('SD_WEBUI_RESTARTING', '1')
print('Restarting UI...')
shared.demo.close()
time.sleep(0.5)
startup_timer.reset()
script_callbacks.app_reload_callback()
startup_timer.record("app reload callback")
script_callbacks.script_unloaded_callback()
startup_timer.record("scripts unloaded callback")
initialize.initialize_rest(reload_script_modules=True)
def api_only():
Thread(target=api_only_worker, daemon=True).start()
def webui():
Thread(target=webui_worker, daemon=True).start()
if __name__ == "__main__":
from modules.shared_cmd_options import cmd_opts
if cmd_opts.nowebui:
api_only()
else:
webui()
main_thread.loop()
|