Spaces:
Configuration error
Configuration error
File size: 20,854 Bytes
7f30a93 c132e32 8a8c958 c132e32 7f30a93 c132e32 8a8c958 7f30a93 8a8c958 7f30a93 c132e32 7f30a93 c132e32 7f30a93 8a8c958 7f30a93 8a8c958 7f30a93 8a8c958 c132e32 7f30a93 c132e32 d809aeb 7f30a93 d809aeb 7f30a93 2d6a1ff 7f30a93 d809aeb 7f30a93 d809aeb 8a8c958 d809aeb 7f30a93 d809aeb 8a8c958 f06506b 8a8c958 c132e32 f06506b 5a185d0 7f30a93 5a185d0 c132e32 f06506b 8a8c958 c132e32 f06506b 5a185d0 f06506b 5a185d0 c132e32 f06506b 86be414 7f30a93 c132e32 7f30a93 f06506b 7f30a93 f06506b 86be414 f06506b 7f30a93 f06506b 86be414 7f30a93 86be414 7f30a93 86be414 f06506b 8a8c958 f06506b 7f30a93 86be414 7f30a93 f06506b 86be414 7f30a93 f06506b 86be414 f06506b 86be414 f06506b 86be414 7f30a93 f06506b c132e32 d809aeb f06506b 7f30a93 f06506b c132e32 f06506b 7f30a93 f06506b 7f30a93 f06506b d809aeb f06506b c132e32 7f30a93 f06506b c132e32 f06506b 7f30a93 c132e32 5a185d0 f06506b 5a185d0 c132e32 f06506b c132e32 f06506b c132e32 5a185d0 f06506b 5a185d0 c132e32 7f30a93 c132e32 |
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 181 182 183 184 185 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 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 |
from signal import signal, SIGKILL
import gradio as gr
from inspect import getfile
import socket
import requests
import os
class Dock:
def __init__(self) -> None:
self.port_map = dict()
for p in range(7860, 7880):
if not self.port_is_connected(p):
self.port_map[p] = True
else:
self.port_map[p] = False
def port_is_connected(self, port : int) -> bool:
"""
@params:
- port : int
@return:
- boolean
check if the port is open with in our localhost
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex(("localhost", port)) == 0
def determinePort(self) -> any:
"""
Take the port_map that was instantiate
in the __init__ and loop though all ports
and check if it the port is available
"""
for port, available in self.port_map.items():
if available == True:
if self.port_is_connected(port): # check if port is in use if so then go to the next one
continue
return port
raise Exception(f'❌ 🔌 {bcolor.BOLD}{bcolor.UNDERLINE}{bcolor.FAIL}All visable ports are used up...Try close some ports {bcolor.ENDC}')
DOCKER_LOCAL_HOST = '0.0.0.0'
DOCKER_PORT = Dock() # Determine the best possible port
def InterLauncher(name, interface, listen=2000, **kwargs):
"""
@params:
- name : string
- interface : gradio.Interface(...)
- listen : int
- **kwargs
Take any gradio interface object
that is created by the gradio
package and send it to the flaks api
"""
port= kwargs["port"] if "port" in kwargs else DOCKER_PORT.determinePort() # determine the next open port is no port is listed **kwargs
try:
# (POST) send the information of the gradio to the flask api
requests.post(f"http://{DOCKER_LOCAL_HOST}:{listen}/api/append/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : "Not Applicable", "name" : name, "kwargs" : kwargs})
except Exception as e:
# If there is an Exception then notify the user and end the method
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The listening api is either not up or you choose the wrong port.🐛 \n {e}")
return
# Launch the gradio application
interface.launch(server_port=port,
server_name=f"{DOCKER_LOCAL_HOST}",
inline= kwargs['inline'] if "inline" in kwargs else False,
share=kwargs['share'] if "share" in kwargs else None,
debug=kwargs['debug'] if "debug" in kwargs else False,
enable_queue=kwargs['enable_queue'] if "enable_queue" in kwargs else None,
max_threads=kwargs['max_threads'] if "max_threads" in kwargs else None,
auth=kwargs['auth'] if "auth" in kwargs else None,
auth_message=kwargs['auth_message'] if "auth_message" in kwargs else None,
prevent_thread_lock=kwargs['prevent_thread_lock'] if "prevent_thread_lock" in kwargs else False,
show_error=kwargs['show_error'] if "show_error" in kwargs else True,
show_tips=kwargs['show_tips'] if "show_tips" in kwargs else False,
height=kwargs['height'] if "height" in kwargs else 500,
width=kwargs['width'] if "width" in kwargs else 900,
encrypt=kwargs['encrypt'] if "encrypt" in kwargs else False,
favicon_path=kwargs['favicon_path'] if "favicon_path" in kwargs else None,
ssl_keyfile=kwargs['ssl_keyfile'] if "ssl_keyfile" in kwargs else None,
ssl_certfile=kwargs['ssl_certfile'] if "ssl_certfile" in kwargs else None,
ssl_keyfile_password=kwargs['ssl_keyfile_password'] if "ssl_keyfile_password" in kwargs else None,
quiet=kwargs['quiet'] if "quiet" in kwargs else False)
try:
# (POST) stop the interface if user hits ctrl+c
# send the information of the gradio to the flask
# api to remove it from the list within the flask api
requests.post(f"http://{DOCKER_LOCAL_HOST}:{ listen }/api/remove/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : 'Not Applicable', "name" : name, "kwargs" : kwargs})
except Exception as e:
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The api either lost connection or was turned off...🐛 \n {e}")
def tabularGradio(funcs, names=[], name="Tabular Temp Name", **kwargs):
"""
takes all gradio Interfaces, and names
from input and launch the gradio.
"""
fn = [fn() for fn in funcs if not isinstance(fn, gr.Interface) and fn.__decorator__ == "__gradio__" ]
if len(names) == 0:
names = [_.__name__ for _ in fn]
port= kwargs["port"] if "port" in kwargs else DOCKER_PORT.determinePort()
# send this to the backend api for it to be read by the react frontend
if 'listen' in kwargs:
try:
# (POST) send the information of the gradio to the flask api
requests.post(f"http://{DOCKER_LOCAL_HOST}:{ kwargs[ 'listen' ] }/api/append/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : 'Not Applicable', "name" : name, "kwargs" : kwargs})
except Exception as e:
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The listening api is either not up or you choose the wrong port.🐛 \n {e}")
return
# provided by gradio a tabularInterface function that take function and names
gr.TabbedInterface(fn, names).launch(server_port=port,
server_name=f"{DOCKER_LOCAL_HOST}",
inline= kwargs['inline'] if "inline" in kwargs else True,
share=kwargs['share'] if "share" in kwargs else None,
debug=kwargs['debug'] if "debug" in kwargs else False,
enable_queue=kwargs['enable_queue'] if "enable_queue" in kwargs else None,
max_threads=kwargs['max_threads'] if "max_threads" in kwargs else None,
auth=kwargs['auth'] if "auth" in kwargs else None,
auth_message=kwargs['auth_message'] if "auth_message" in kwargs else None,
prevent_thread_lock=kwargs['prevent_thread_lock'] if "prevent_thread_lock" in kwargs else False,
show_error=kwargs['show_error'] if "show_error" in kwargs else True,
show_tips=kwargs['show_tips'] if "show_tips" in kwargs else False,
height=kwargs['height'] if "height" in kwargs else 500,
width=kwargs['width'] if "width" in kwargs else 900,
encrypt=kwargs['encrypt'] if "encrypt" in kwargs else False,
favicon_path=kwargs['favicon_path'] if "favicon_path" in kwargs else None,
ssl_keyfile=kwargs['ssl_keyfile'] if "ssl_keyfile" in kwargs else None,
ssl_certfile=kwargs['ssl_certfile'] if "ssl_certfile" in kwargs else None,
ssl_keyfile_password=kwargs['ssl_keyfile_password'] if "ssl_keyfile_password" in kwargs else None,
quiet=kwargs['quiet'] if "quiet" in kwargs else False)
# Ctrl+C that ends the process and then continue the code which will remove from the api
if 'listen' in kwargs:
try:
requests.post(f"http://{DOCKER_LOCAL_HOST}:{ kwargs[ 'listen' ] }/api/remove/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : 'Not Applicable', "name" : name, "kwargs" : kwargs})
except Exception as e:
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The api either lost connection or was turned off...🐛 \n {e}")
return
def register(inputs, outputs, examples=None, **kwargs):
"""
Decorator that is appended to a function either within a class or not
and output either an interface or inputs and outputs for later processing
to launch either to Gradio-Flow or just Gradio
"""
def register_gradio(func):
def decorator(*args, **wargs):
# if the output is a list then there should be equal or more then 1
if type(outputs) is list:
assert len(outputs) >= 1, f"❌ {bcolor.BOLD}{bcolor.FAIL}you have no outputs 🤨... {str(type(outputs))} {bcolor.ENDC}"
fn_name = func.__name__ # name of the function
# if there exist the self within the arguments and thats the first argument then this must be a class function
if 'self' in func.__code__.co_varnames and func.__code__.co_varnames[0] == 'self' and fn_name in dir(args[0]):
"""
given the decorator is on a class then
initialize a registered_gradio_functons
if not already initialize.
"""
# if the inputs is a list then inputs list should equal the arugments list
if type(inputs) is list:
assert len(inputs) == func.__code__.co_argcount - 1, f"❌ {bcolor.BOLD}{bcolor.FAIL}inputs should have the same length as arguments{bcolor.ENDC}"
try:
self = args[0]
self.registered_gradio_functons
except AttributeError:
self.registered_gradio_functons = dict() # if registered_gradio_functons does not exist then create it
# if the function name is not within the registered_gradio_functons then register it within the registered_gradio_functons
if not fn_name in self.registered_gradio_functons:
self.registered_gradio_functons[fn_name] = dict(inputs=inputs,
outputs=outputs,
examples=examples,
cache_examples=kwargs['cache_examples'] if "cache_examples" in kwargs else None,
examples_per_page=kwargs['examples_per_page'] if "examples_per_page" in kwargs else 10,
interpretation=kwargs['interpretation'] if "interpretation" in kwargs else None,
num_shap=kwargs['num_shap'] if "num_shap" in kwargs else 2.0,
title=kwargs['title'] if "title" in kwargs else None,
article=kwargs['article'] if "article" in kwargs else None,
thumbnail=kwargs['thumbnail'] if "thumbnail" in kwargs else None,
css=kwargs['css'] if "css" in kwargs else None,
live=kwargs['live'] if "live" in kwargs else False,
allow_flagging=kwargs['allow_flagging'] if "allow_flagging" in kwargs else None,
theme=kwargs['theme'] if "theme" in kwargs else 'default', )
# if the argument are within the function when it's called then give me the output of the function
# giving the user the ability to use the function if necessary
if len(args[1:]) == (func.__code__.co_argcount - 1):
return func(*args, **wargs)
# return nothing if the arguments are not within it cause if the arguments do not exist it will give a error
return None
else :
"""
the function is not a class function
"""
# if the inputs is a list then inputs list should equal the arugments list
if type(inputs) is list:
assert len(inputs) == func.__code__.co_argcount, f"❌ {bcolor.BOLD}{bcolor.FAIL}inputs should have the same length as arguments{bcolor.ENDC}"
# if the arguments within the functions are inputed then just return the output
if len(args) == (func.__code__.co_argcount):
return func(*args, **wargs)
# if there is nothing in the arugumrnt then return the gradio interface
return gr.Interface(fn=func,
inputs=inputs,
outputs=outputs,
examples=examples,
cache_examples=kwargs['cache_examples'] if "cache_examples" in kwargs else None,
examples_per_page=kwargs['examples_per_page'] if "examples_per_page" in kwargs else 10,
interpretation=kwargs['interpretation'] if "interpretation" in kwargs else None,
num_shap=kwargs['num_shap'] if "num_shap" in kwargs else 2.0,
title=kwargs['title'] if "title" in kwargs else None,
article=kwargs['article'] if "article" in kwargs else None,
thumbnail=kwargs['thumbnail'] if "thumbnail" in kwargs else None,
css=kwargs['css'] if "css" in kwargs else None,
live=kwargs['live'] if "live" in kwargs else False,
allow_flagging=kwargs['allow_flagging'] if "allow_flagging" in kwargs else None,
theme='default',
)
decorator.__decorator__ = "__gradio__" # siginture to tell any function that need to know that function is a registed gradio application
return decorator
return register_gradio
def GradioModule(cls):
class Decorator:
def __init__(self, *args, **kwargs) -> None:
self.__cls__ = cls(*args, **kwargs)
self.__get_funcs_attr()
self.interface = self.__compile()
def get_funcs_names(self):
"""
Get all name for each function
"""
assert self.get_registered_map() != None, "this is not possible..."
return [ name for name in self.get_registered_map().keys()]
def get_registered_map(self):
"""
Get all registered functions
"""
assert self.__cls__.registered_gradio_functons != None, "what happen!!!!"
return self.__cls__.registered_gradio_functons
def __get_funcs_attr(self):
"""
Get all the function that are registered
"""
for func in dir(self.__cls__):
fn = getattr(self.__cls__, func, None)
if callable(fn) and not func.startswith("__") and "__decorator__" in dir(fn) and fn.__decorator__ == "__gradio__":
fn()
def __compile(self):
"""
Initialize all the function
within the class that are registeed
"""
demos, names = [], []
for func, param in self.get_registered_map().items(): # loop though the registered function and append it to the TabularInterface
names.append(func)
try:
demos.append(gr.Interface(fn=getattr(self.__cls__, func, None), **param))
except Exception as e :
raise e
print(f"{bcolor.OKBLUE}COMPLETED: {bcolor.ENDC}All functions are mapped, and ready to launch 🚀",
"\n===========================================================\n")
return gr.TabbedInterface(demos, names)
def launch(self, **kwargs):
"""
@params:
**kwargs
Take the tabular interface and send it to the api if
'listen' is within the kwargs and launch the gradio interface
then when the gradio stops then remove it from the api
"""
port= kwargs["port"] if "port" in kwargs else DOCKER_PORT.determinePort()
if 'listen' in kwargs:
try:
requests.post(f"http://{DOCKER_LOCAL_HOST}:{ kwargs[ 'listen' ] }/api/append/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : getfile(self.__cls__.__class__), "name" : self.__cls__.__class__.__name__, "kwargs" : kwargs})
except Exception:
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The listening api is either not up or you choose the wrong port.🐛")
return
self.interface.launch(server_port=port,
server_name=f"{DOCKER_LOCAL_HOST}",
inline= kwargs['inline'] if "inline" in kwargs else None,
share=kwargs['share'] if "share" in kwargs else None,
debug=kwargs['debug'] if "debug" in kwargs else False,
enable_queue=kwargs['enable_queue'] if "enable_queue" in kwargs else None,
max_threads=kwargs['max_threads'] if "max_threads" in kwargs else None,
auth=kwargs['auth'] if "auth" in kwargs else None,
auth_message=kwargs['auth_message'] if "auth_message" in kwargs else None,
prevent_thread_lock=kwargs['prevent_thread_lock'] if "prevent_thread_lock" in kwargs else False,
show_error=kwargs['show_error'] if "show_error" in kwargs else True,
show_tips=kwargs['show_tips'] if "show_tips" in kwargs else False,
height=kwargs['height'] if "height" in kwargs else 500,
width=kwargs['width'] if "width" in kwargs else 900,
encrypt=kwargs['encrypt'] if "encrypt" in kwargs else False,
favicon_path=kwargs['favicon_path'] if "favicon_path" in kwargs else None,
ssl_keyfile=kwargs['ssl_keyfile'] if "ssl_keyfile" in kwargs else None,
ssl_certfile=kwargs['ssl_certfile'] if "ssl_certfile" in kwargs else None,
ssl_keyfile_password=kwargs['ssl_keyfile_password'] if "ssl_keyfile_password" in kwargs else None,
quiet=kwargs['quiet'] if "quiet" in kwargs else False)
if 'listen' in kwargs:
try:
requests.post(f"http://{DOCKER_LOCAL_HOST}:{ kwargs[ 'listen' ] }/api/remove/port", json={"port" : port, "host" : f'http://localhost:{port}', "file" : getfile(self.__cls__.__class__), "name" : self.__cls__.__class__.__name__, "kwargs" : kwargs})
except Exception:
print(f"**{bcolor.BOLD}{bcolor.FAIL}CONNECTION ERROR{bcolor.ENDC}** 🐛The api either lost connection or was turned off...🐛")
return
return Decorator
# console colour changer
class bcolor:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKCYAN = '\033[96m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
|