ws-terminal / main.py
memex-in's picture
Create main.py
84c35be verified
import asyncio, gradio as gr
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
# WebSocket terminal
@app.websocket("/ws")
async def ws_terminal(ws: WebSocket):
await ws.accept()
await ws.send_text("Connected to terminal\n")
try:
while True:
cmd = await ws.receive_text()
proc = await asyncio.create_subprocess_shell(cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE)
out, err = await proc.communicate()
msg = (out + err).decode()
await ws.send_text(msg)
except WebSocketDisconnect:
print("Client disconnected")
# Gradio UI
JS = """
<script>
const ws = new WebSocket(`ws://${location.host}/ws`);
ws.onmessage = e => {
const o = document.getElementById("out");
o.value += e.data + "\\n";
o.scrollTop = o.scrollHeight;
};
function sendCmd(e) {
if (e.key === "Enter") {
ws.send(e.target.value);
e.target.value = "";
}
}
</script>
<textarea id="out" rows="20" style="width:100%;" readonly></textarea>
<input id="inp" style="width:100%;" placeholder="cmd…" onkeydown="sendCmd(event)">
"""
with gr.Blocks() as demo:
gr.Markdown("## 🖥️ Interactive Terminal")
gr.HTML(JS)
# Launch Gradio on port 7861
gr_app = gr.mount_gradio_app(app, demo, path="/gradio", server_port=7861)