File size: 5,817 Bytes
5f7f889 | 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 | import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from typing import List
app = FastAPI()
# --- Connection Manager ---
class ConnectionManager:
def __init__(self):
# Store active connections
self.macos_connections: List[WebSocket] = []
self.linux_connections: List[WebSocket] = []
async def connect_macos(self, websocket: WebSocket):
await websocket.accept()
self.macos_connections.append(websocket)
print(f"Client connected to macOS. Total: {len(self.macos_connections)}")
async def connect_linux(self, websocket: WebSocket):
await websocket.accept()
self.linux_connections.append(websocket)
print(f"Client connected to Linux. Total: {len(self.linux_connections)}")
def disconnect_macos(self, websocket: WebSocket):
if websocket in self.macos_connections:
self.macos_connections.remove(websocket)
print("Client disconnected from macOS")
def disconnect_linux(self, websocket: WebSocket):
if websocket in self.linux_connections:
self.linux_connections.remove(websocket)
print("Client disconnected from Linux")
async def forward_to_linux_clients(self, message: str):
# Forward message to all Linux clients
# Iterate over a copy of the list to avoid issues if a client disconnects mid-loop
for connection in self.linux_connections[:]:
try:
await connection.send_text(message)
except Exception as e:
print(f"Error sending to linux client: {e}")
async def forward_to_macos_clients(self, message: str):
# Forward message to all macOS clients
for connection in self.macos_connections[:]:
try:
await connection.send_text(message)
except Exception as e:
print(f"Error sending to macos client: {e}")
manager = ConnectionManager()
# --- WebSocket Endpoints ---
@app.websocket("/macos")
async def websocket_macos(websocket: WebSocket):
await manager.connect_macos(websocket)
try:
while True:
# Wait for data from macOS client
data = await websocket.receive_text()
print(f"Received from macOS: {data}")
# Forward exact same thing to Linux clients
await manager.forward_to_linux_clients(data)
except WebSocketDisconnect:
manager.disconnect_macos(websocket)
except Exception as e:
print(f"Error in macOS socket: {e}")
manager.disconnect_macos(websocket)
@app.websocket("/linux")
async def websocket_linux(websocket: WebSocket):
await manager.connect_linux(websocket)
try:
while True:
# Wait for data from Linux client
data = await websocket.receive_text()
print(f"Received from Linux: {data}")
# Forward exact same thing to macOS clients
await manager.forward_to_macos_clients(data)
except WebSocketDisconnect:
manager.disconnect_linux(websocket)
except Exception as e:
print(f"Error in Linux socket: {e}")
manager.disconnect_linux(websocket)
# --- Simple UI for testing (Optional) ---
@app.get("/")
async def get():
return HTMLResponse("""
<!DOCTYPE html>
<html>
<head>
<title>Bridge Tester</title>
</head>
<body>
<h1>OS Bridge Tester</h1>
<div style="display:flex; gap: 20px;">
<div style="border:1px solid #ccc; padding:10px;">
<h2>Pretend to be macOS</h2>
<button onclick="connect('macos')">Connect to /macos</button>
<input type="text" id="msgMac" placeholder="Message to Linux" />
<button onclick="send('macos')">Send</button>
<ul id="logMac"></ul>
</div>
<div style="border:1px solid #ccc; padding:10px;">
<h2>Pretend to be Linux</h2>
<button onclick="connect('linux')">Connect to /linux</button>
<input type="text" id="msgLin" placeholder="Message to macOS" />
<button onclick="send('linux')">Send</button>
<ul id="logLin"></ul>
</div>
</div>
<script>
var wsMac = null;
var wsLin = null;
function connect(os) {
var ws = new WebSocket(((window.location.protocol === "https:") ? "wss://" : "ws://") + window.location.host + "/" + os);
ws.onopen = function() { appendLog(os, "Connected!"); };
ws.onmessage = function(event) { appendLog(os, "Received: " + event.data); };
if(os === 'macos') wsMac = ws;
else wsLin = ws;
}
function send(os) {
var input = document.getElementById(os === 'macos' ? 'msgMac' : 'msgLin');
var ws = os === 'macos' ? wsMac : wsLin;
if(ws) ws.send(input.value);
input.value = '';
}
function appendLog(os, msg) {
var ul = document.getElementById(os === 'macos' ? 'logMac' : 'logLin');
var li = document.createElement("li");
li.appendChild(document.createTextNode(msg));
ul.appendChild(li);
}
</script>
</body>
</html>
""")
# Entry point for Hugging Face (Starts on port 7860)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860) |