backend/src/resources/module.py

import gradio as gr
from inspect import getfile, isclass, getmro

import socket
import random
import requests

class Dock:

    def __init__(self) -> None:
            self.num_ports = 20
            self.port_range = (7860, 7880)

    def portConnection(self, port : int):
            s = socket.socket(
                socket.AF_INET, socket.SOCK_STREAM)  
            result = s.connect_ex(("localhost", port))
            if result == 0: return True
            return False

    def determinePort(self, max_trial_count=10):
            trial_count = 0 
            while trial_count <= max_trial_count:
                port=random.randint(*self.port_range)
                if not self.portConnection(port):
                    return port
                trial_count += 1
            raise Exception('Exceeded Max Trial count without finding port')


DOCKER_LOCAL_HOST = '0.0.0.0'
DOCKER_PORT = Dock()

def tabularGradio(funcs, names, name="Tabular Temp Name", **kwargs):
    #print([fn.__name__ for fn in funcs])
    assert len(funcs) == len(names), f"{bcolor.BOLD}{bcolor.FAIL}🐛 something went wrong!!! The function you appended dose not match the lenght of the names{bcolor.ENDC}"
    # assert all([fn == "wrap" for fn in funcs]), f"{bcolor().BOLD}{bcolor().FAIL}not all of these are decorated with the right decorator{bcolor().ENDC}"
    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" : '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
    
    gr.TabbedInterface(funcs, 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)
    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):
    def register_gradio(func):
        def wrap(*args, **wargs):                    
            fn_name = func.__name__ 

            if 'self' in func.__code__.co_varnames and func.__name__ in dir(args[0]):
                
                """
                given the decorator is on a class then
                initialize a registered_gradio_functons
                if not already initialize.
                """
                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:
                    print("✨Initializing Class Functions...✨\n")
                    self.registered_gradio_functons = dict()
                
                

                if not fn_name in self.registered_gradio_functons:
                    self.registered_gradio_functons[fn_name] = dict(inputs=inputs, outputs=outputs, examples=examples)
                
                if len(args[1:]) == (func.__code__.co_argcount - 1):
                    return func(*args, **wargs) 
            else :
                """
                the function is not a class function
                """
                assert len(inputs) == func.__code__.co_argcount, f"❌ {bcolor.BOLD}{bcolor.FAIL}inputs should have the same length as arguments{bcolor.ENDC}"

                if len(args) == (func.__code__.co_argcount):
                    return func(*args, **wargs)

                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', 
                                    )

                return None
        return wrap
    return register_gradio

def GradioModule(cls):
    class Decorator:

        def __init__(self) -> None:
            self.cls = cls()

        def get_funcs(self):
            return [func for func in dir(self.cls) if not func.startswith("__") and type(getattr(self.cls, func, None)) == type(self.get_funcs) ]

        def compile(self, **kwargs):
            print("Just putting on the finishing touches... 🔧🧰")
            for func in self.get_funcs():
                this = getattr(self.cls, func, None)
                if this.__name__ == "wrap":
                    this()

            demos, names = [], []
            for func, param in self.get_registered_gradio_functons().items():                
                names.append(func)
                demos.append(gr.Interface(fn=getattr(self.cls, func, None),
                                            inputs=param['inputs'],
                                            outputs=param['outputs'],
                                            examples=param['examples'],
                                            cache_examples=kwargs['cache_examples'] if "cache_examples" in kwargs else None,
                                            examples_per_page=kwargs['cache_examples'] if "cache_examples" 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', 
                                            ))
                print(f"{func}....{bcolor.BOLD}{bcolor.OKGREEN} done {bcolor.ENDC}")

            print("\nHappy Visualizing... 🚀")
            return gr.TabbedInterface(demos, names)
            
        def get_registered_gradio_functons(self):
            try:
                self.cls.registered_gradio_functons
            except AttributeError:
                return None
            return self.cls.registered_gradio_functons
        

        def run(self, **kwargs):
            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.compile(live=kwargs[ 'live' ] if "live" in kwargs else False,
                         allow_flagging=kwargs[ 'allow_flagging' ] if "allow_flagging" in kwargs else None,
                         cache_examples=kwargs['cache_examples'] if "cache_examples" in kwargs else None,
                         examples_per_page=kwargs['cache_examples'] if "cache_examples" 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,
                         theme=kwargs['theme'] if "theme" in kwargs else None, 
                         ).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) 
            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



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'