import asyncio
from starlette.applications import Starlette
from starlette.endpoints import WebSocketEndpoint, HTTPEndpoint
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute, Mount
from starlette.websockets import WebSocket
from starlette.types import Message, Receive, Scope, Send
# -------------------- gradio ------------------------
import gradio as gr
async def name_fn(name):
await mbase.add_msg(f'gradio_name: {name}')
return "Name: " + name + "!"
io = gr.Interface(fn=name_fn, inputs=gr.Textbox(lines=2, placeholder="Name Here..."), outputs="text")
gradio_app = gr.routes.App.create_app(io)
# -------------------- shiny ------------------------
import shiny
from shiny import App, render, ui, reactive
import numpy as np
import matplotlib.pyplot as plt
app_ui = ui.page_fluid(
ui.input_slider("n", "Number of bins", 1, 100, 20),
ui.output_plot("plot", width='30%', height='120px'),
)
def server(input, output, session):
@output
@render.plot(alt="A histogram")
async def plot():
await mbase.add_msg(f'shiny_bin: {input.n()}')
x = 100 + 15 * np.random.randn(120)
fig, ax = plt.subplots()
ax.hist(x, input.n(), density=True)
return fig
app_shiny = App(ui = app_ui, server = server)
# -------------------- main + msg store and dispatch ------------------------
ws_list = []
class MsgBase():
""" Store and send messages to clients """
msg_history = ['start logging..',]
async def add_msg(self, msg):
self.msg_history.insert(0, msg)
if len(self.msg_history) > 15:
self.msg_history.pop()
for ws in ws_list:
await ws.send_text(msg)
async def send_all_msg(self, ws):
""" Restore all messages in new page or reloaded page"""
ws_list.append(ws)
for msg in self.msg_history:
await ws.send_text(msg)
mbase = MsgBase()
html = """
Shiny and Gradio Test App
Shiny App
Gradio App
"""
class Homepage(HTTPEndpoint):
async def get(self, request):
return HTMLResponse(html)
class Echo(WebSocketEndpoint):
async def on_connect(self, websocket: WebSocket) -> None:
await websocket.accept()
await mbase.send_all_msg(websocket)
async def on_disconnect(self, websocket: WebSocket, close_code: int) -> None:
ws_list.remove(websocket)
routes = [
Route("/", Homepage),
WebSocketRoute("/ws", Echo),
Mount('/shiny', app=app_shiny),
Mount('/gradio', app=gradio_app)
]
app = Starlette(routes=routes)
# as a main work loop..
@app.on_event("startup")
async def startup_event():
asyncio.create_task(dispatch_task())
async def dispatch_task():
""" background task to dispatch autonomous events """
for i in range(20):
await mbase.add_msg(f'(main loop) I am work!({i})')
await asyncio.sleep(8)