# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.
"""A base class for graph visualization servers.
See GraphServer for more information
"""
import json
import inspect
import webbrowser
from pathlib import Path
from collections.abc import Sequence
from Karana.WebUI import HttpWsServer
from ._datatypes import NetworkGraph, Button, JsonType, Edge
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)
headers = [("Content-Type", "text/html")]
self.serveData("/standalone.html", html_text, headers=headers)
[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 setSelection(self, ids: Sequence[int | str]):
"""Set which node and/or edge ids are selected on the frontends.
Parameters
----------
node_id: int | str
Id of the node to select
"""
self.broadcastMessage(self._buildSetSelectionMsg(ids))
[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 addEdge(self, edge: Edge):
"""Add a new edge.
This immediately broadcasts the change to clients.
Parameters
----------
edge: Edge
The new edge to add
"""
# Add the edge to the saved graph. This is needed to keep our
# server-side graph consistent for any later connecting
# clients.
self.graph.edges.append(edge)
# Send a message to all clients to update their frontends
self.broadcastMessage(self._buildAddEdgeMsg(edge))
[docs]
def removeEdge(self, edge: Edge | int | str):
"""Remove any edges with the given id.
This immediately broadcasts the change to clients.
Parameters
----------
edge: Edge | int | str
Edge or Edge id to be removed
"""
# We need an id, so either get it from the edge or assume an id
# was given
if isinstance(edge, Edge):
edge_id = edge.id
else:
edge_id = edge
# Filter out any edges in the graph with matching id. This is
# needed to keep our server-side graph consistent for any later
# connecting clients.
self.graph.edges = [e for e in self.graph.edges if e.id != edge_id]
# Send a message to all clients to update their frontends
self.broadcastMessage(self._buildRemoveEdgeMsg(edge_id))
[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")
def _buildSetSelectionMsg(self, ids: Sequence[int | str]) -> bytes:
"""Build a 'set_selection' message helper method."""
payload = {"ids": ids}
return json.dumps({"type": "set_selection", "payload": payload}).encode("utf-8")
def _buildAddEdgeMsg(self, edge: Edge) -> bytes:
"""Build an 'add_edge' message helper method."""
return json.dumps({"type": "add_edge", "payload": edge.toDict()}).encode("utf-8")
def _buildRemoveEdgeMsg(self, edge_id: int | str) -> bytes:
"""Build a 'remove_edge' message helper method."""
return json.dumps({"type": "remove_edge", "payload": edge_id}).encode("utf-8")
