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 Karana.WebUI import HttpWsServer
from ._datatypes import NetworkGraph, Button, JsonType
from ._standalone import buildStandaloneHtml
__all__ = ["GraphServer"]
class GraphServer(HttpWsServer):
"""A generic graph visualization server.
This builds on HttpWsServer and uses the vis.js library to provide
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,
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.
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
super().__init__(port)
static_path = Path(__file__).resolve().parent / "static"
self.serveFile("/", static_path / "index.html")
self.serveFile("/static/index.html", static_path / "index.html")
self.serveFile("/static/icon.jpg", static_path / "icon.jpg")
self.serveFile("/static/script.js", static_path / "script.js")
self.serveFile("/static/style.css", static_path / "style.css")
self.serveFile("/static/vis-network.min.js", static_path / "vis-network.min.js")
self.serveFile("/static/vis-network.min.js.map", static_path / "vis-network.min.js.map")
self.setOnConnect(self.onConnect)
self.setOnMessage(self.onMessage)
self.setOnDisconnect(self.onDisconnect)
self.graph = graph or NetworkGraph()
self.buttons = {}
if buttons:
for button in buttons:
self.addButton(button)
# TODO: make this dynamically resolved at time of request
html_text = buildStandaloneHtml(self.graph)
self.serveData("/standalone.html", html_text, "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:
url = f"{self.getUrl()}/standalone.html"
else:
url = self.getUrl()
webbrowser.open(url)
[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.broadcastMessage(self._buildUpdateGraphMsg())
[docs]
def onConnect(self, client_id: int):
"""Handle a new client connecting.
We initialize the graph and UI elements of the new client.
Parameters
----------
client_id: int
Value identifying the connected client
"""
# What to do when a client first connects
message = self._buildUpdateGraphMsg()
self.sendMessage(message, client_id)
for button in self.buttons.values():
message = self._buildAddButtonMsg(button)
self.sendMessage(message, client_id)
[docs]
def onMessage(self, message: str, client_id):
"""Handle receiving a websocket client message.
Based on the message type we delegate to the appropriate handler
method.
Parameters
----------
message: str
The message text as a plain unparsed string
client_id: int
Value identifying the client
"""
# 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":
self.onClickNode(client_id, payload)
elif msg_type == "edge_click":
self.onClickEdge(client_id, payload)
elif msg_type == "button_click":
button_id = payload["id"]
context = payload["context"]
self._onButtonPress(client_id, button_id=button_id, context=context)
except Exception as err:
print(err)
[docs]
def onDisconnect(self, client_id: int):
"""Handle a websocket client disconnecting.
There's nothing extra we need to do so this is a no-op.
Parameters
----------
client_id: int
Value identifying the client
"""
[docs]
def onClickNode(self, client_id: int, node_id: int | str):
"""When a user clicks on a node, this method is called.
Parameters
----------
client_id: int
Value identifying the client
node_id : int | str
The node clicked on.
"""
pass
[docs]
def onClickEdge(self, client_id: int, edge_id: int | str):
"""When a user clicks on an edge, this method is called.
Parameters
----------
client_id: int
Value identifying the client
edge_id : int | str
The edge clicked on.
"""
pass
def _onButtonPress(self, client_id: int, button_id: str, context: JsonType):
"""Route a button press to the correct callback helper method.
Parameters
----------
client_id: int
Value identifying the client
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()
def _buildUpdateGraphMsg(self) -> bytes:
"""Build a 'graph_update' message helper method."""
return json.dumps({"type": "graph_update", "payload": self.graph.toDict()}).encode("utf-8")
def _buildAddButtonMsg(self, button: Button) -> bytes:
"""Build an 'add_button' message helper method."""
payload = button.toDict()
del payload["callback"]
return json.dumps({"type": "add_button", "payload": payload}).encode("utf-8")
