Source code for Karana.KUtils.visjs._server
"""A base class for graph visualization servers.
See GraphServer for more information
"""
import json
import inspect
import webbrowser
from pathlib import Path
from fastapi import WebSocket
from fastapi.responses import Response
from ._baseserver import HybridServerBase
from ._datatypes import NetworkGraph, Button, JsonType
from ._standalone import buildStandaloneHtml
__all__ = ["GraphServer"]
class GraphServer(HybridServerBase):
"""A generic graph visualization server.
This builds on HybridServerBase and uses the vis.js library to
provide a server for visualizing and interacting with graph
topologies in the web browser. The server can be specialized with
frontend buttons and event handlers that trigger server-side
callbacks. Additionally the server can broadcast to clients changes
to the graph topology and appearance via a live websocket
connection.
"""
def __init__(
self,
graph: NetworkGraph | None = None,
*,
port=8765,
autorun: bool = True,
log_level: str = "warning",
buttons: list[Button] | None = None,
):
"""Create a GraphServer instance.
Parameters
----------
graph: NetworkGraph | None
If not None, the initial graph to send to clients
port: int
The port to bind to. Must not already be in use. The port
number 0 picks an arbitrary free port. Defaults to 8765.
autorun: bool
If True, automatically start the server. Otherwise the user
must call the run method to start the server. Defaults to
True.
log_level: str
The logging verbosity level passed to uvicorn. Should be a
string such as "info" or "warning". Defaults to "warning".
buttons: list[Button] | None
A list of buttons to add to the frontend UI. Includes
server-side callbacks associated with each button.
"""
# Serve the required static files found in a `static` directory
# next to this .py file
static_path = Path(__file__).resolve().parent / "static"
super().__init__(static_path=static_path, port=port, autorun=autorun, log_level=log_level)
self.graph = graph or NetworkGraph()
self.buttons = {}
if buttons:
for button in buttons:
self.addButton(button)
[docs]
def setupRoutes(self):
"""Add http request handlers.
In additional to everything added in
HybridServerBase.setupRoutes, this adds the `/standalone.html`
route, which serves a fully self-contained html file showing the
graph. This file can be saved and later reopened in a web
browser for offline viewing.
This may be overridden to add more routes, but in this case it is
recommended to call `super().setupRoutes()` so that the routes
added by this class are preserved.
"""
super().setupRoutes()
@self.app.get("/standalone.html")
async def serveStandalone():
html_text = buildStandaloneHtml(self.graph)
return Response(
content=html_text,
media_type="text/html",
)
[docs]
def launchLocalClient(self, standalone: bool = False):
"""Open a client in a browser tab local to the server.
Parameters
----------
standalone: bool
If True, instead open the `/standalone.html` URL, which
serves a self-contained HTML file that is missing some
features but can be saved and later opened for offline
viewing. Defaults to False.
"""
if standalone:
webbrowser.open(f"{self.url}/standalone.html")
else:
super().launchLocalClient()
[docs]
def updateClientGraphs(self):
"""Update clients to use the current graph.
This must be called after modifying or replacing self.graph for
the modifications to show up in connected clients.
"""
self.broadcast(self._buildUpdateGraphMsg())
[docs]
async def onConnect(self, websocket: WebSocket):
"""Handle a new client connecting.
We initialize the graph and UI elements of the new client.
Parameters
----------
websocket: WebSocket
The websocket client which just connected
"""
# What to do when a client first connects
message = self._buildUpdateGraphMsg()
await websocket.send_text(message)
# Add custom buttons to the new client
for button in self.buttons.values():
await websocket.send_text(self._buildAddButtonMsg(button))
[docs]
async def onMessage(self, websocket: WebSocket, message: str):
"""Handle receiving a websocket client message.
Based on the message type we delegate to the appropriate handler
method.
Parameters
----------
websocket: WebSocket
The websocket client which sent the message
message: str
The message text as a plain unparsed string
"""
# What to do when we get a message from a client
try:
parsed = json.loads(message)
msg_type = parsed["type"]
payload = parsed.get("payload", None)
if msg_type == "node_click":
await self.onClickNode(websocket, payload)
elif msg_type == "edge_click":
await self.onClickEdge(websocket, payload)
elif msg_type == "button_click":
button_id = payload["id"]
context = payload["context"]
await self._onButtonPress(websocket, button_id=button_id, context=context)
except Exception as err:
print(err)
[docs]
async def onDisconnect(self, websocket: WebSocket):
"""Handle a websocket client disconnecting.
There's nothing extra we need to do so this is a no-op.
Parameters
----------
websocket: WebSocket
The websocket client which just disconnected
"""
[docs]
async def onClickNode(self, websocket: WebSocket, node_id: int | str):
"""When a user clicks on a node, this method is called.
Parameters
----------
websocket : WebSocket
The associated websocket.
node_id : int | str
The node clicked on.
"""
pass
[docs]
async def onClickEdge(self, websocket: WebSocket, edge_id: int | str):
"""When a user clicks on an edge, this method is called.
Parameters
----------
websocket : WebSocket
The associated websocket.
edge_id : int | str
The edge clicked on.
"""
pass
async def _onButtonPress(self, websocket: WebSocket, button_id: str, context: JsonType):
"""Route a button press to the correct callback helper method.
Parameters
----------
websocket: WebSocket
The websocket client where the button was pressed
button_id: str
Unique id for the button that was pressed
context: FrontendContext
Information about the frontend state
"""
callback = self.buttons[button_id].callback
# Allow callback to be no-arugment or pass frontend context if it takes
# one argument
signature = inspect.signature(callback)
if len(signature.parameters) > 0:
result = callback(context)
else:
result = callback()
# Handle the case where the given callback is async
if inspect.isawaitable(result):
await result
def _buildUpdateGraphMsg(self) -> str:
"""Build a 'graph_update' message helper method."""
return json.dumps({"type": "graph_update", "payload": self.graph.toDict()})
def _buildAddButtonMsg(self, button: Button) -> str:
"""Build an 'add_button' message helper method."""
payload = button.toDict()
del payload["callback"]
return json.dumps({"type": "add_button", "payload": payload})
