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("""
Bridge Tester
OS Bridge Tester
""")
# Entry point for Hugging Face (Starts on port 7860)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860)