Source code for Karana.KUtils.visjs._server
"""This module defines 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,
):
"""GraphServer constructor
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 overriden 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 serve_standalone():
html_text = buildStandaloneHtml(self.graph)
return Response(
content=html_text,
media_type="text/html",
)
[docs]
def launchLocalClient(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 un-parsed 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):
# What to do when the user clicks on a node
pass
[docs]
async def onClickEdge(self, websocket: WebSocket, edge_id: int | str):
# What to do when the user clicks on an edge
pass
async def _onButtonPress(self, websocket: WebSocket, button_id: str, context: JsonType):
"""Helper to route a button press to the correct callback
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:
"""Helper to build a 'graph_update' message"""
return json.dumps({"type": "graph_update", "payload": self.graph.to_dict()})
def _buildAddButtonMsg(self, button: Button) -> str:
"""Helper to build an 'add_button' message"""
payload = button.to_dict()
del payload["callback"]
return json.dumps({"type": "add_button", "payload": payload})
