Spaces:
Running
Running
| from toolbox import get_conf | |
| from toolbox import set_conf | |
| from toolbox import set_multi_conf | |
| from toolbox import get_plugin_handle | |
| from toolbox import get_plugin_default_kwargs | |
| from toolbox import get_chat_handle | |
| from toolbox import get_chat_default_kwargs | |
| from functools import wraps | |
| import sys | |
| import os | |
| def chat_to_markdown_str(chat): | |
| result = "" | |
| for i, cc in enumerate(chat): | |
| result += f'\n\n{cc[0]}\n\n{cc[1]}' | |
| if i != len(chat)-1: | |
| result += '\n\n---' | |
| return result | |
| def silence_stdout(func): | |
| def wrapper(*args, **kwargs): | |
| _original_stdout = sys.stdout | |
| sys.stdout = open(os.devnull, 'w') | |
| sys.stdout.reconfigure(encoding='utf-8') | |
| for q in func(*args, **kwargs): | |
| sys.stdout = _original_stdout | |
| yield q | |
| sys.stdout = open(os.devnull, 'w') | |
| sys.stdout.reconfigure(encoding='utf-8') | |
| sys.stdout.close() | |
| sys.stdout = _original_stdout | |
| return wrapper | |
| def silence_stdout_fn(func): | |
| def wrapper(*args, **kwargs): | |
| _original_stdout = sys.stdout | |
| sys.stdout = open(os.devnull, 'w') | |
| sys.stdout.reconfigure(encoding='utf-8') | |
| result = func(*args, **kwargs) | |
| sys.stdout.close() | |
| sys.stdout = _original_stdout | |
| return result | |
| return wrapper | |
| class VoidTerminal(): | |
| def __init__(self) -> None: | |
| pass | |
| vt = VoidTerminal() | |
| vt.get_conf = silence_stdout_fn(get_conf) | |
| vt.set_conf = silence_stdout_fn(set_conf) | |
| vt.set_multi_conf = silence_stdout_fn(set_multi_conf) | |
| vt.get_plugin_handle = silence_stdout_fn(get_plugin_handle) | |
| vt.get_plugin_default_kwargs = silence_stdout_fn(get_plugin_default_kwargs) | |
| vt.get_chat_handle = silence_stdout_fn(get_chat_handle) | |
| vt.get_chat_default_kwargs = silence_stdout_fn(get_chat_default_kwargs) | |
| vt.chat_to_markdown_str = chat_to_markdown_str | |
| proxies, WEB_PORT, LLM_MODEL, CONCURRENT_COUNT, AUTHENTICATION, CHATBOT_HEIGHT, LAYOUT, API_KEY = \ | |
| vt.get_conf('proxies', 'WEB_PORT', 'LLM_MODEL', 'CONCURRENT_COUNT', 'AUTHENTICATION', 'CHATBOT_HEIGHT', 'LAYOUT', 'API_KEY') | |
| def plugin_test(main_input, plugin, advanced_arg=None): | |
| from rich.live import Live | |
| from rich.markdown import Markdown | |
| vt.set_conf(key="API_KEY", value=API_KEY) | |
| vt.set_conf(key="LLM_MODEL", value=LLM_MODEL) | |
| plugin = vt.get_plugin_handle(plugin) | |
| plugin_kwargs = vt.get_plugin_default_kwargs() | |
| plugin_kwargs['main_input'] = main_input | |
| if advanced_arg is not None: | |
| plugin_kwargs['plugin_kwargs'] = advanced_arg | |
| my_working_plugin = silence_stdout(plugin)(**plugin_kwargs) | |
| with Live(Markdown(""), auto_refresh=False) as live: | |
| for cookies, chat, hist, msg in my_working_plugin: | |
| md_str = vt.chat_to_markdown_str(chat) | |
| md = Markdown(md_str) | |
| live.update(md, refresh=True) |