# 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.
"""Classes and functions used to stream data to a DashApp.
This module contains classes and functions used to stream data to a DashApp. These are useful for
plotting data while a simulation is running. The DashApp is a useful format, as it allows one to
view the results from a web browser, and that browser does not have to be on the same machine as
the machine generating the data. This is particularly useful when running a simulation on a
remote machine without an X session.
"""
import os
from Karana.Core import debug
from threading import Thread, Event, Lock
from werkzeug.serving import make_server, WSGIRequestHandler
from typing import TYPE_CHECKING, Any, cast, overload
from dash import Dash, html, Input, Output, dcc
from pathlib import Path
import plotly.graph_objects as go
from dash_extensions import WebSocket
import websockets
import asyncio
import socket
import struct
from json import dumps
if TYPE_CHECKING:
from Karana.KUtils import SinglePlotData, PlotData
else:
from ._KUtils_Py import SinglePlotData, PlotData
class _DashServer:
"""_DashSever is an internal class used to start/stop the server.
Users should not create instances of this themselves. We do this in
two classes to avoid reference cycles. Users need only delete the
DashApp instance, and the server should shutdown automatically.
"""
def __init__(self, title: str = "DataPlotter", host: str = "0.0.0.0", port: int = 8050):
"""Initialize the DashServer.
Parameters
----------
host: str
The host to host the server on.
port: int
The port to use for the server.
"""
self.host = host
if self.host == "0.0.0.0":
self.hostname = os.getenv("KARANA_WEBUI_ADDRESS", "localhost")
else:
self.hostname = self.host
if port != 0:
port_offset_str = os.getenv("KARANA_PORT_OFFSET", None)
port_offset = int(port_offset_str) if port_offset_str else 0
port += port_offset
asset_folder = Path(__file__).parents[0].joinpath("assets")
self._websocket_loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._websocket_loop)
self.port: int
self.app = Dash(__name__, title=title, assets_folder=str(asset_folder))
self.update_clients = set()
self.refresh_clients = set()
self._startServer(port)
# Store the figures on the DashServer as well. If we keep them on the main, then we can't cleanup.
self.figs: list[go.Figure] = []
self.gs: list[dcc.Graph] = []
self.fig_labels: list[str] = []
# Counter for the figures
self.figure_counter: int = 0
# Append-only histories avoid copying complete Plotly traces for every simulation sample.
self.plot_histories: list[dict[str, Any]] = []
self.history_lock = Lock()
def _runServer(self, port: int, event: Event):
"""Run the server.
This is the method run by the _server_thread.
"""
class SilentHandler(WSGIRequestHandler):
def log(self, type, message, *args):
pass # Do nothing — suppress all logs
try:
self._server = make_server(
self.host, port, self.app.server, request_handler=SilentHandler
)
# If port 0 was passed in, then make_server will select an available port.
# If non-zero, then this should be the same as port anyway. Use this so
# any printed messages have the real port number.
except Exception:
event.set()
raise
self.port = self._server.port
event.set()
self._server.serve_forever()
def _runWebsocket(self, event: Event):
"""Run the websocket.
This is the method run by the _websocket_thread.
"""
async def handler(websocket):
# Only browsers use WebSockets. Simulation data arrives through framed loopback TCP.
if websocket.request.path == "/refresh":
self.refresh_clients.add(websocket)
# A newly connected browser must request a full-history figure snapshot.
await websocket.send(dumps({"msg": "refresh plots"}))
else:
self.update_clients.add(websocket)
debug(lambda: "Client connected")
try:
async for _ in websocket:
# Browser connections are receive-only, but consuming frames keeps them alive.
pass # pragma: no cover - Browser clients do not normally send messages.
except (
websockets.exceptions.ConnectionClosed
): # pragma: no cover - Exclude browser disconnect timing from coverage.
pass
finally:
debug(lambda: "Client disconnected")
if websocket.request.path == "/refresh":
self.refresh_clients.discard(websocket)
else:
self.update_clients.discard(websocket)
self._websocket_loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._websocket_loop)
self._websocket_stop = asyncio.Event()
self.websocket_port = asyncio.Future()
async def startServer():
server = await websockets.serve(handler, "0.0.0.0", 0)
self.websocket_port.set_result(server.sockets[0].getsockname()[1])
event.set()
debug(lambda: f"Starting WebSocket server on port {self.websocket_port.result()}...")
await self._websocket_stop.wait()
server.close()
await server.wait_closed()
debug(lambda: f"WebSocket server on port {self.websocket_port.result()} closed")
self._websocket_loop.run_until_complete(startServer())
self._websocket_loop.stop()
def _startServer(self, port: int):
"""Start the server.
This creates a thread and adds _run to it.
"""
if not self.isRunning():
# We use daemon threads so atexit functions can run without waiting for these threads to finish.
# Use events so we know when the ports are set
e1 = Event()
e2 = Event()
self._server_thread = Thread(target=self._runServer, args=(port, e1))
self._server_thread.daemon = True
self._server_thread.start()
self._websocket_thread = Thread(target=self._runWebsocket, args=(e2,))
self._websocket_thread.daemon = True
self._websocket_thread.start()
e1.wait()
e2.wait()
print(f"Dash app started at http://{self.hostname}:{self.port}")
def _stopServer(self):
"""Stop the server and join the associated thread."""
if self._websocket_loop and not self._websocket_loop.is_closed():
self._websocket_loop.call_soon_threadsafe(lambda: self._websocket_stop.set())
if self._server:
self._server.shutdown()
self._server = None
print("Dash server shutting down...")
if self._server_thread:
self._server_thread.join()
if self._websocket_thread:
self._websocket_thread.join()
async def _sendWebsocketMessage(self, message: str, path="/update"):
"""Async function to ping the websocket."""
to_remove = set()
if path == "/update":
clients = self.update_clients
elif path == "/refresh":
clients = self.refresh_clients
else:
raise ValueError("path not recognized.")
for ws in clients.copy():
try:
await ws.send(message)
except (
websockets.exceptions.ConnectionClosed
): # pragma: no cover - Exclude from code coverage, as this won't run in testing
debug(lambda: "Client disconnected before send")
to_remove.add(ws)
clients.difference_update(to_remove)
def pingWebsocket(self):
"""Ping the websocket.
Pinging the websocket will also update the plots.
"""
message = {"msg": "update graph!"}
asyncio.run_coroutine_threadsafe(
self._sendWebsocketMessage(dumps(message)), self._websocket_loop
)
def sendWebsocketMessage(self, message: str, path="/update"):
"""Ping the websocket.
Pinging the websocket will also update the plots.
"""
asyncio.run_coroutine_threadsafe(
self._sendWebsocketMessage(message, path=path), self._websocket_loop
)
def isRunning(self) -> bool:
"""Return a boolean that indicates whether the server is running or not."""
return hasattr(self, "_server_thread") and self._server_thread.is_alive()
def __del__(
self,
): # pragma: no cover - Normal reference cycles keep this from being run. We keep it around in case users ever trigger a strange scenario where this is removed (in which case it should shut down the server).
"""Stop the server if it is running."""
self._stopServer()
class _DashAppBase:
"""Base class for the DashApp* classes."""
def __init__(
self,
title: str = "DataPlotter",
host: str = "0.0.0.0",
port: int = 8050,
):
"""Initialize the DashApp.
This starts a server on the provided host and port. This server should shutdown
automatically once this instance is deleted.
Parameters
----------
title: str
The title of the DashApp webpage.
host: str
The host to host the server on.
port: int
The port to use for the server.
"""
self._server = _DashServer(host=host, port=port, title=title)
app = self._server.app
app.callback
self.url = f"http://{self._server.hostname}:{self._server.websocket_port.result()}/update"
app.layout = [
html.Header(
[
html.H1(children=title, style={"textAlign": "center", "flex-grow": "1"}),
html.Button(
r"☀️",
title="Toggle light/dark theme",
id="theme-toggle",
className="toggle-button",
),
],
style={
"display": "flex",
"justify-content": "space-between",
"align-items": "center",
"margin-bottom": "20px",
"padding-bottom": "10px",
},
),
dcc.Store(id="update-data", data=0),
WebSocket(url=self.url, id="ws"),
]
app.clientside_callback(
r"""
function(n_clicks) {
const body = document.body;
const button = document.getElementById('theme-toggle');
const current = body.getAttribute('data-theme') || 'light';
if (current === 'light') {
body.setAttribute('data-theme', 'dark');
button.textContent = '\u{1F319}';
} else {
body.setAttribute('data-theme', 'light');
button.textContent = '\u2600\uFE0F';
}
return '';
}
""",
Output("theme-toggle", "n_clicks"),
Input("theme-toggle", "n_clicks"),
)
def printConnectionInfo(self):
"""Print a message about how to connect to the server."""
http_url = f"http://{self._server.hostname}:{self._server.port}"
lines = [
f"[DashApp] Web server is running on port {self._server.port}",
"You may be able to connect in your browser at:",
f"\t\033[1m{http_url}\033[0m", # ANSI bold
]
print("\n".join(lines))
def __del__(self):
"""Delete method.
This stops the server.'
"""
if hasattr(self, "_server"):
self._server._stopServer()
[docs]
class DashAppStatic(_DashAppBase):
"""Use this class to create a DashApp servers up static figures.
These figures can be changed manually using the figs property. However, they are not designed
to be updated regularly. For that, please use the DashApp class.
"""
def __init__(
self,
figs: list[go.Figure],
title: str = "DataPlotter",
host: str = "0.0.0.0",
port: int = 8050,
):
"""Initialize the DashApp.
This starts a server on the provided host and port. This server should shutdown
automatically once this instance is deleted.
Parameters
----------
title: str
The title of the DashApp webpage.
host: str
The host to host the server on.
port: int
The port to use for the server.
"""
super().__init__(title=title, host=host, port=port)
self.figs = figs
self._createPlots(self._server, self._server.app)
@property
def figs(self) -> list[go.Figure]:
"""Figures shown by the DashApp."""
return self._server.figs
@figs.setter
def figs(self, figs: list[go.Figure]):
self._server.figs = figs
self._server.pingWebsocket()
# Using a static method here so we don't create reference cycles that prevent d from being deleted.
@staticmethod
def _createPlots(server, app):
gs = [
dcc.Graph(id=fig.layout.title.text or str(k), figure=fig)
for k, fig in enumerate(server.figs)
]
app.layout.append(html.Div(id="plots-div", children=gs))
@app.callback(Output("plots-div", "children"), Input("ws", "message"))
def updatePlots(_):
gs = [
dcc.Graph(id=fig.layout.title.text or str(k), figure=fig)
for k, fig in enumerate(server.figs)
]
return gs
def _snapshotFigures(server) -> list[go.Figure]:
"""Populate figures from append-only history for a full browser refresh.
Parameters
----------
server : _DashServer
Server containing the figure definitions and live trace histories.
Returns
-------
list[plotly.graph_objects.Figure]
Figures populated with every sample received by the server.
"""
with server.history_lock:
for fig, plot_history in zip(server.figs, server.plot_histories):
for trace, y_history in zip(fig.data, plot_history["y"]):
# Every trace in a plot shares one X history, avoiding duplicate time storage.
trace.x = plot_history["x"]
trace.y = y_history
return server.figs
def _replaceFigures(server, figures: list[go.Figure]):
"""Replace figures and rebuild append-only history from their trace data.
Parameters
----------
server : _DashServer
Server whose figures and histories are replaced.
figures : list[plotly.graph_objects.Figure]
New figures, including any existing trace samples that must be retained.
"""
with server.history_lock:
server.figs = figures
server.plot_histories = []
for figure in figures:
# Plotly figures conventionally share X data; retain the first trace's X values once.
first_trace = figure.data[0] if figure.data else None
server.plot_histories.append(
{
"x": list(first_trace.x) if first_trace and first_trace.x is not None else [],
"y": [list(trace.y) if trace.y is not None else [] for trace in figure.data],
}
)
def _decodeSamples(payload: bytes | memoryview) -> list[tuple[float, list[float]]]:
"""Decode one binary numeric sample frame.
Parameters
----------
payload : bytes or memoryview
Frame body after the one-byte binary message discriminator.
Returns
-------
list[tuple[float, list[float]]]
Per-plot X values and flattened trace values.
"""
offset = 0
plot_count = struct.unpack_from("!I", payload, offset)[0]
offset += 4
samples = []
for _ in range(plot_count):
x_value = struct.unpack_from("!d", payload, offset)[0]
offset += 8
value_count = struct.unpack_from("!I", payload, offset)[0]
offset += 4
values = list(struct.unpack_from(f"!{value_count}d", payload, offset))
offset += value_count * 8
samples.append((x_value, values))
if offset != len(payload):
raise ValueError("Binary plot frame contains trailing data")
return samples
def _receiveExactly(connection: socket.socket, buffer: bytearray, size: int) -> memoryview | None:
"""Receive an exact number of bytes into a reusable buffer.
Parameters
----------
connection : socket.socket
Connected producer socket.
buffer : bytearray
Reusable destination with capacity for at least ``size`` bytes.
size : int
Number of bytes required.
Returns
-------
memoryview or None
View of the received bytes, or ``None`` when the peer closes before the frame is complete.
"""
view = memoryview(buffer)[:size]
received = 0
while received < size:
count = connection.recv_into(view[received:], size - received)
if count == 0:
return None
received += count
return view
class _DashAppClient(_DashAppBase):
"""Use this class to create a DashApp that you can stream data to.
This class will automatically create empty plots based on the `SinglePlotData`s provided. In addition,
it will start a websocket and http server for clients to connect to view the plots. Calling
`update` will update the data of the plots based on the data provided in the `SinglePlotData`s.
"""
def __init__(
self,
data: list[dict[str, Any]],
title: str = "DataPlotter",
host: str = "0.0.0.0",
port: int = 8050,
batch_size: int = 16,
max_latency: float = 0.05,
):
"""Initialize the DashApp.
This starts a server on the provided host and port. This server should shutdown
automatically once this instance is deleted.
Parameters
----------
data : list[SinglePlotData]
Data to be plotted and update functions.
title: str
The title of the DashApp webpage.
host: str
The host to host the server on.
port: int
The port to use for the server.
batch_size : int
Number of simulation updates to combine before immediately sending a browser batch.
max_latency : float
Maximum seconds a partial browser batch may wait for another simulation update.
"""
super().__init__(title=title, host=host, port=port)
# Pending browser updates are grouped by plot and flushed by size or latency.
self._pending_plots: dict[str, dict[str, Any]] = {}
self._pending_sample_count = 0
self._flush_handle: asyncio.TimerHandle | None = None
self.batch_size = batch_size
self.max_latency = max_latency
# Add the plots
for d in data:
# Using underscore method here to avoid calling the refresh method, since it doesn't exist yet.
self._addPlot(d)
# Add extra HTML elements
self._server.app.layout.append(
html.Div(
[
WebSocket(
url=f"http://{self._server.hostname}:{self._server.websocket_port.result()}/refresh",
id="ws-refresh",
),
html.Div(
id="ws-url-div",
children=f"ws://{self._server.hostname}:{self._server.websocket_port.result()}/update",
style={"display": "none"},
),
html.Div(id="plots-div"),
]
)
)
# Use the static method to create callbacks. This avoids tying up self, which leads to reference cycles
# that cause issues when tearing things down.
self._createPlots(self._server, self._server.app)
@staticmethod
def _createPlots(server, app):
# This callback refreshes the plots entirely
@app.callback(Output("plots-div", "children"), Input("ws-refresh", "message"))
def refreshPlots(msg):
# Materialize complete Plotly traces only for a new/refreshing browser.
server.figs = _snapshotFigures(server)
server.gs = [
dcc.Graph(id=label, figure=fig)
for label, fig in zip(server.fig_labels, server.figs)
]
return server.gs
# Refresh the plots with the initial data
server.sendWebsocketMessage(dumps({"msg": "update graph!"}), path="/refresh")
def update(self, inc_data: list[tuple[float, list[float]]]):
"""Update the plots.
Data is appended to the complete server-side history and to a compact browser batch. The
batch is sent when either ``batch_size`` updates accumulate or ``max_latency`` expires.
Parameters
----------
inc_data : list[tuple[float, list[float]]]
Per-plot X value and flattened trace values decoded from the binary TCP frame.
"""
with self._server.history_lock:
for plot_index, (plot_id, data) in enumerate(zip(self._server.fig_labels, inc_data)):
x_data, y_values = data
history = self._server.plot_histories[plot_index]
history["x"].append(x_data)
for y_history, y_value in zip(history["y"], y_values):
y_history.append(y_value)
pending = self._pending_plots.get(plot_id)
if pending is None:
pending = {"id": plot_id, "x": [], "y": [[] for _ in y_values]}
self._pending_plots[plot_id] = pending
pending["x"].append(x_data)
for trace_values, y_value in zip(pending["y"], y_values):
trace_values.append(y_value)
self._pending_sample_count += 1
if self._pending_sample_count >= self.batch_size:
self._flushPendingData()
elif self._flush_handle is None:
# The deadline guarantees a partial batch is sent even if no more samples arrive.
self._flush_handle = self._server._websocket_loop.call_later(
self.max_latency, self._flushPendingData
)
def setBatchSize(self, batch_size: int):
"""Change the number of updates that triggers an immediate batch flush.
Parameters
----------
batch_size : int
New positive batch size. A pending batch is flushed immediately when it already
contains at least this many updates.
"""
self.batch_size = batch_size
if self._pending_sample_count >= self.batch_size:
self._flushPendingData()
def setMaxLatency(self, max_latency: float):
"""Change how long a partial batch may wait before being sent.
Parameters
----------
max_latency : float
New non-negative latency in seconds. An outstanding deadline restarts from the time
this method is called.
"""
self.max_latency = max_latency
if self._flush_handle is not None:
# Restart an outstanding deadline using the newly requested latency.
self._flush_handle.cancel()
self._flush_handle = self._server._websocket_loop.call_later(
self.max_latency, self._flushPendingData
)
def _flushPendingData(self):
"""Send pending samples as one compact, plot-oriented browser message.
Notes
-----
Pending buffers are swapped before transmission so samples received during the WebSocket
send are accumulated in the next batch.
"""
if self._flush_handle is not None:
self._flush_handle.cancel()
self._flush_handle = None
if not self._pending_plots:
return
# Swap buffers before sending so newly arriving samples join the next batch.
payload = {"plots": list(self._pending_plots.values())}
self._pending_plots = {}
self._pending_sample_count = 0
self._server.sendWebsocketMessage(dumps(payload, separators=(",", ":")))
def addPlot(self, data: dict[str, Any]):
"""Add a new plot the window using the associated plot data.
Parameters
----------
data : dict[str, Any]
Example data associated with the plot.
"""
self._flushPendingData()
self._addPlot(data)
self._server.sendWebsocketMessage(dumps({"msg": "new plot added."}), path="/refresh")
def removePlot(self, index: int):
"""Remove the plot at the associated index.
The index corresponds to the figure in the slot of self._server.figs.
Parameters
----------
index : int
The index of the plot to remove.
"""
self._flushPendingData()
with self._server.history_lock:
self._server.figs.pop(index)
self._server.fig_labels.pop(index)
self._server.gs.pop(index)
self._server.plot_histories.pop(index)
self._server.sendWebsocketMessage(dumps({"msg": "removing plot"}), path="/refresh")
def _addPlot(self, data: dict[str, Any]):
"""Add a new plot the window using the associated plot data.
This does not call the plot refresh. Use the addPlot method for that.
Parameters
----------
data : dict[str, Any]
The data associated with updating the plot.
"""
# Create the figure based on the incoming data
fig = go.Figure()
for j, data_len in data["y_data"].items():
if data_len == 1:
fig.add_trace(go.Scatter(x=[], y=[], mode="lines+markers", name=j))
else:
for i in range(data_len):
fig.add_trace(go.Scatter(x=[], y=[], mode="lines+markers", name=f"{j}[{i}]"))
fig.update_layout(
template="plotly_dark", xaxis_title=data["x_data_name"], title=data["title"]
)
with self._server.history_lock:
# Publish all structural state atomically with respect to full-history snapshots.
label = f"karana-plot-{self._server.figure_counter}"
# Add this to the server's list of figures
self._server.figs.append(fig)
# Store each trace in append-friendly lists; figures are snapshots, not live storage.
self._server.plot_histories.append({"x": [], "y": [[] for _ in range(len(fig.data))]})
# Add the label to the server's list of figure labels
self._server.fig_labels.append(label)
# Add this to the server's Graph objects
self._server.gs.append(dcc.Graph(id=label, figure=fig))
# Increment the figure counter
self._server.figure_counter += 1
if TYPE_CHECKING:
from multiprocessing import Event, Queue
from multiprocessing.synchronize import Event as Evt
def _startProcess(
shutdown: "Evt",
change_plot: "Evt",
q: "Queue",
data: dict[str, Any],
title: str,
host: str,
port: int,
batch_size: int,
max_latency: float,
):
"""Start function for the DashApp multiprocessing process.
Parameters
----------
shutdown : Evt
An event to indicate shutdown.
change_plot : Evt
An event to indicate we are changing something fundamental about the plots.
q : Queue
A multiprocessing Queue to exchange data.
data : list[SinglePlotData]
The data to use to create the plots.
title : str
The title of the DashApp.
host : str
The host for the DashApp.
port : int
The port to server the DashApp on.
batch_size : int
Number of simulation samples that triggers an immediate browser batch.
max_latency : float
Maximum seconds a partial browser batch may wait before being sent.
"""
from json import loads
# Bind loopback TCP before creating the parent-side PlotData connection. Unlike Unix-domain
# sockets, this framed transport behaves the same on supported macOS and Linux systems.
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(("127.0.0.1", 0))
listener.listen(1)
data_port = listener.getsockname()[1]
client = _DashAppClient(data, title, host, port, batch_size, max_latency)
# Send data back to the parent process
q.put(f"http://{client._server.hostname}:{client._server.port}")
q.put("127.0.0.1")
q.put(data_port)
def processMessage(message):
"""Apply one decoded producer message on the WebSocket event-loop thread.
Parameters
----------
message : list or dict
Plot samples or a control message decoded from the framed TCP payload.
"""
if isinstance(message, list):
client.update(message)
elif isinstance(message, dict) and "REMOVE_PLOT" in message:
client.removePlot(message["REMOVE_PLOT"])
change_plot.set()
elif isinstance(message, dict) and "ADD_PLOT" in message:
client.addPlot(message["ADD_PLOT"])
change_plot.set()
elif isinstance(message, dict) and "GET_FIGS" in message:
q.put(_snapshotFigures(client._server))
elif isinstance(message, dict) and "SET_FIGS" in message:
client._flushPendingData()
_replaceFigures(client._server, q.get())
client._server.sendWebsocketMessage(dumps({"msg": "updating plot"}), path="/refresh")
change_plot.set()
elif isinstance(message, dict) and "SET_BATCH_SIZE" in message:
client.setBatchSize(message["SET_BATCH_SIZE"])
elif isinstance(message, dict) and "SET_MAX_LATENCY" in message:
client.setMaxLatency(message["SET_MAX_LATENCY"])
def receiveData():
"""Receive framed binary or JSON messages and schedule Dash event-loop handling.
Notes
-----
This function runs on a dedicated thread because socket reads are blocking.
"""
connection, _ = listener.accept()
listener.close()
with connection:
connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
header_buffer = bytearray(4)
payload_buffer = bytearray()
while True:
header = _receiveExactly(connection, header_buffer, 4)
if header is None:
break
message_size = struct.unpack("!I", header)[0]
# Grow only when a larger frame arrives; smaller frames reuse existing capacity.
if message_size > len(payload_buffer):
payload_buffer = bytearray(message_size)
payload = _receiveExactly(connection, payload_buffer, message_size)
if payload is None:
break
message_type = payload[0]
if message_type == ord("D"):
message = _decodeSamples(payload[1:])
elif message_type == ord("J"):
# JSON controls are infrequent, so materializing their small payload is fine.
message = loads(payload[1:].tobytes())
else:
raise ValueError("Unknown plot transport message type")
client._server._websocket_loop.call_soon_threadsafe(processMessage, message)
# Release views before a future iteration replaces or grows their bytearrays.
del header
del payload
# Keep blocking socket reads away from both the WSGI and browser WebSocket threads.
receiver_thread = Thread(target=receiveData, daemon=True)
receiver_thread.start()
# Wait until we get the signal from the parent to shutdown
shutdown.wait()
# Closing PlotData in the parent closes the TCP peer and lets the receiver finish.
receiver_thread.join(timeout=10.0)
# Explicitly stop the server. del just removes things from the local scope, but
# the process may exit after that since interpolation is techincally done (while
# logic from __del__ is still running). This ensures the server is shutdown
# every time before closing.
client._server._stopServer()
[docs]
class DashApp(PlotData):
"""Use this class to create a DashApp that you can stream data to.
This class will automatically create empty plots based on the `SinglePlotData`s provided. In addition,
it will start a websocket and http server for clients to connect to view the plots. Calling
`update` will update the data of the plots based on the data provided in the `SinglePlotData`s.
"""
def __init__(
self,
data: list[SinglePlotData],
title: str = "DataPlotter",
host: str = "0.0.0.0",
port: int = 8050,
):
"""Initialize the DashApp.
This starts a server on the provided host and port. This server should shutdown
automatically once this instance is deleted.
Parameters
----------
data : list[SinglePlotData]
Data to be plotted and update functions.
title: str
The title of the DashApp webpage.
host: str
The host to host the server on.
port: int
The port to use for the server.
"""
from multiprocessing import Event, Process, Queue
# Set per-instance defaults without exposing additional constructor arguments. Control
# messages remain disabled until the parent-side TCP transport has connected.
self._transport_ready = False
self.batch_size = 32
self.max_latency = 0.05
self.shutdown_event = shutdown = Event()
self.change_plot = change_plot = Event()
q = self._mp_queue = Queue()
# Get data for the initial plots
dat = []
for d in data:
dark = {
"title": d.title,
"x_data_name": d.x_data_name,
"x_data": 1,
"y_data": {k: v for k, v in zip(d.y_data_names, d.y_data_sizes)},
}
dat.append(dark)
self.p = Process(
target=_startProcess,
args=(
shutdown,
change_plot,
q,
dat,
title,
host,
port,
self._batch_size,
self._max_latency,
),
)
self.p.daemon = True
self.p.start()
# Wait on the values from the queue
# This is the server url
self.url: str = q.get()
# This is the internal TCP transport hostname/port.
hostname = cast(str, q.get())
p = str(q.get())
self._transport_url = f"tcp://{hostname}:{p}"
# Call the parent constructor
super().__init__(data, hostname, p, "/data")
self._transport_ready = True
@property
def batch_size(self) -> int:
"""Number of simulation updates that triggers an immediate browser batch.
Returns
-------
int
Current positive batch size.
"""
return self._batch_size
@batch_size.setter
def batch_size(self, value: int):
"""Set the number of updates that triggers an immediate browser batch.
Parameters
----------
value : int
Positive number of updates per batch.
Raises
------
ValueError
If ``value`` is a boolean, is not an integer, or is less than one.
"""
if not isinstance(value, int) or isinstance(value, bool) or value < 1:
raise ValueError("batch_size must be an integer greater than zero")
self._batch_size = value
if self._transport_ready:
# Propagate changes made after construction to the child plotting process.
self.sendMessage(dumps({"SET_BATCH_SIZE": value}))
@property
def max_latency(self) -> float:
"""Maximum time a partial browser batch may wait before being sent.
Returns
-------
float
Current maximum latency in seconds.
"""
return self._max_latency
@max_latency.setter
def max_latency(self, value: float):
"""Set the maximum time a partial browser batch may wait.
Parameters
----------
value : float
Non-negative latency in seconds.
Raises
------
ValueError
If ``value`` is a boolean, is not numeric, or is negative.
"""
if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0:
raise ValueError("max_latency must be a non-negative number")
self._max_latency = float(value)
if self._transport_ready:
# Propagate changes made after construction to the child plotting process.
self.sendMessage(dumps({"SET_MAX_LATENCY": self._max_latency}))
@overload
def removePlot(self, title: str, /):
"""Remove the plot with the associated title.
This will only work if there is one and only one title associated
with the plot.
"""
...
@overload
def removePlot(self, index: int, /):
"""Remove the plot at the associated index.
The index corresponds to the figure in the slot of self._server.figs.
Parameters
----------
index : int
The index of the plot to remove.
"""
...
[docs]
def removePlot(self, arg: int | str, /): # pyright: ignore - override with different signature on purpuse
"""Remove the plot based on the value of arg.
See overloads for more details.
"""
index = None
if isinstance(arg, str):
for k, d in enumerate(self.plot_fns):
if d.title == arg:
if index is not None:
raise ValueError(f"More than one plot has the title {arg}")
else:
index = k
if index is None:
raise ValueError(f"No match found for title {arg}")
else:
index = arg
self.change_plot.clear()
super().removePlot(index)
self.change_plot.wait()
[docs]
def addPlot(self, plot_data: SinglePlotData):
"""Add a plot.
Parameters
----------
plot_fns : SinglePlotData
The plotting functions associated with the new plot.
"""
self.change_plot.clear()
super().addPlot(plot_data)
self.change_plot.wait()
@property
def figs(self) -> list[go.Figure]:
"""Get the figures associated with this DashApp."""
self.sendMessage(dumps({"GET_FIGS": 1}))
return self._mp_queue.get()
@figs.setter
def figs(self, figs: list[go.Figure]):
"""Set the figures associated with this DashApp."""
self.change_plot.clear()
self._mp_queue.put(figs)
self.sendMessage(dumps({"SET_FIGS": 1}))
self.change_plot.wait()
[docs]
def __del__(self):
"""Shut everything down."""
self.shutdown()
if self.p.is_alive():
# If p is still alive, try to shut it down cleanly.
self.shutdown_event.set()
if os.environ.get("DTEST_RUNNING", False):
# If we are running tests in parallel, then don't set a timeout.
# The strain of the tests means this may take a little while if the CPUs
# are maxed.
self.p.join()
else:
self.p.join(timeout=10.0)
