Model Serving Monitor#
A live operations view for a simulated fraud-scoring model. Pick a traffic profile and go live to watch request rate, latency, error rate and feature drift respond. Threshold alerts fire and resolve automatically, and scaling replicas resolves a traffic surge. Click a recent request to inspect its payload.
All telemetry comes from a seeded pseudo-random simulator in this session. The dashboard starts paused.
import json
import math
from collections import deque
import numpy as np
import pandas as pd
import param
from bokeh.models import (
ColumnDataSource, DatetimeTickFormatter, HoverTool, Span,
)
from bokeh.plotting import figure
import panel as pn
pn.extension("echarts", "tabulator", notifications=True, sizing_mode="stretch_width")
Configuration#
Traffic profiles, alert rules and the capacity model are plain data, so the simulator, the alert logic and the About tab all read from the same definitions.
SEED = 7
WINDOW = 600 # ticks of simulated history kept in memory; one tick is one simulated second
TREND_WINDOW = 120
TABLE_ROWS = 120
TREND_DIGITS = {"rps": 0, "p95": 0, "error_rate": 2, "drift": 3}
REPLICA_RPS = 90
START = pd.Timestamp("2026-09-29 12:00:00")
INDIGO, TEAL, AMBER, RED = "#6366F1", "#14B8A6", "#F59E0B", "#EF4444"
PROFILES = {
"Steady": dict(rps=180, latency=38, errors=0.003, drift=0.04, blurb="Normal weekday traffic with healthy headroom."),
"Traffic surge": dict(rps=560, latency=38, errors=0.003, drift=0.05, blurb="A marketing push triples demand. Add replicas to absorb it."),
"Latency regression": dict(rps=190, latency=115, errors=0.004, drift=0.04, blurb="A slow feature-store lookup inflates every request."),
"Data drift": dict(rps=200, latency=40, errors=0.006, drift=0.34, blurb="A new merchant segment shifts the input distribution."),
"Upstream outage": dict(rps=175, latency=55, errors=0.07, drift=0.05, blurb="A dependency starts failing and requests return 5xx."),
}
RULES = [
dict(key="p95", name="p95 latency above SLO", title="p95 latency > 250 ms", threshold=250, unit="ms", severity="warning", fmt="{:.0f}"),
dict(key="error_rate", name="Error rate above budget", title="Error rate > 2 %", threshold=2.0, unit="%", severity="error", fmt="{:.2f}"),
dict(key="drift", name="Feature drift detected", title="Drift PSI > 0.2", threshold=0.2, unit="PSI", severity="warning", fmt="{:.3f}"),
dict(key="saturation", name="Replica saturation", title="Saturation > 90 %", threshold=90, unit="%", severity="info", fmt="{:.0f}"),
]
HEALTH_ICONS = {"Healthy": "check_circle", "Degraded": "warning", "Critical": "error"}
HEALTH_COLORS = {"Healthy": "success", "Degraded": "warning", "Critical": "error"}
FIRE_AFTER, RESOLVE_AFTER = 3, 5 # consecutive ticks, so a single noisy sample neither fires nor resolves an alert
ENDPOINTS = ["/v1/score", "/v1/score/batch", "/v1/explain"]
MERCHANTS = ["grocery", "travel", "electronics", "fuel", "restaurants", "gaming"]
COUNTRIES = ["DE", "FR", "NL", "ES", "IT", "PL"]
Simulator#
Each tick is one simulated second. Metrics ease towards the selected profile, and latency and errors grow once load exceeds the replicas’ capacity.
class ServingSimulator:
"""Deterministic pseudo-random telemetry for a model endpoint."""
def __init__(self, seed=SEED):
self.rng = np.random.default_rng(seed)
self.tick = 0
self.state = {k: v for k, v in PROFILES["Steady"].items() if k in ("rps", "latency", "errors", "drift")}
self.request_id = 10_000
def step(self, profile, replicas):
rng, target = self.rng, PROFILES[profile]
# Metrics ease towards the profile target so incidents ramp up and recover rather than switching instantly.
for key, value in self.state.items():
self.state[key] = value + (target[key] - value) * 0.08
rps = max(self.state["rps"] * (1 + 0.07 * math.sin(self.tick / 40)) * rng.lognormal(0, 0.04), 1)
util = rps / (replicas * REPLICA_RPS)
queueing = 1 + 6 * max(0, util - 0.7) ** 2
p50 = self.state["latency"] * queueing * rng.lognormal(0, 0.05)
p95 = p50 * (2.3 + 0.35 * abs(rng.normal()))
errors = self.state["errors"] + 0.2 * max(0, util - 0.95) + abs(rng.normal(0, 0.0008))
drift = max(self.state["drift"] + rng.normal(0, 0.008), 0)
metrics = dict(
time=START + pd.Timedelta(seconds=self.tick), rps=rps, p50=p50, p95=p95,
error_rate=min(errors, 1) * 100, drift=drift, saturation=min(util * 100, 100),
)
self.tick += 1
return metrics, [self._request(metrics) for _ in range(rng.poisson(1.5) + 1)]
def _request(self, metrics):
rng = self.rng
self.request_id += 1
drift = metrics["drift"]
err = metrics["error_rate"] / 100
status = int(rng.choice([200, 500, 503], p=[1 - err, err * 0.6, err * 0.4]))
latency = float(metrics["p50"] * rng.lognormal(0, 0.45))
features = {
"amount_eur": round(float(rng.lognormal(3.6 + 2.5 * drift, 0.8)), 2),
"merchant_category": str(rng.choice(MERCHANTS[: 4 + int(drift > 0.15) * 2])),
"country": str(rng.choice(COUNTRIES)),
"card_present": bool(rng.random() > 0.35 + drift),
"device_age_days": int(rng.gamma(2, 120)),
}
score = float(np.clip(rng.beta(1.2, 9) + drift * 0.6, 0, 1))
decision = "decline" if score > 0.7 else "review" if score > 0.4 else "approve"
payload = {
"request_id": f"req-{self.request_id}",
"model": "fraud-scorer",
"model_version": "3.2.1",
"endpoint": (endpoint := str(rng.choice(ENDPOINTS, p=[0.8, 0.15, 0.05]))),
"features": features,
"response": (
{"status": status, "score": round(score, 4), "decision": decision} if status == 200
else {"status": status, "error": "upstream feature store timeout"}
),
"latency_ms": round(latency, 1),
}
row = dict(
time=metrics["time"].strftime("%H:%M:%S"), request_id=payload["request_id"], endpoint=endpoint,
status=status, latency_ms=round(latency), score=round(score, 3) if status == 200 else math.nan,
decision=decision if status == 200 else "error",
)
return row, payload
Pure helpers#
Everything the dashboard displays is derived from parameter values by these functions. The panes bind to them once, so a tick only changes parameters and never rebuilds a component.
def availability(recent):
"""Request-weighted availability over a window of metric samples."""
return round(100 - float((recent.error_rate * recent.rps).sum() / recent.rps.sum()), 2)
def health_for(active):
severities = {record["rule"]["severity"] for record in active.values()}
return "Critical" if "error" in severities else "Degraded" if severities - {"info"} else "Healthy"
def signal_severity(active, rule):
return rule["severity"] if rule["key"] in active else "success"
def signal_variant(active, key):
return "filled" if key in active else "outlined"
def signal_text(active, key):
return f"Firing since {active[key]['started']:%H:%M:%S}" if key in active else "Within threshold"
def payload_summary(payload):
if not payload:
return "Click a request to inspect its payload."
status = payload["response"]["status"]
return f"**{payload['request_id']}** on `{payload['endpoint']}` returned **{status}** in **{payload['latency_ms']:.0f} ms**."
def payload_severity(payload):
if not payload:
return "info"
return "success" if payload["response"]["status"] == 200 else "error"
def payload_json(payload):
return json.dumps(payload, indent=2) if payload else ""
def profile_blurb(profile):
return PROFILES[profile]["blurb"]
def profile_severity(profile):
return "info" if profile == "Steady" else "warning"
def capacity_text(replicas):
return f"≈ {replicas * REPLICA_RPS} req/s before queueing starts to hurt latency."
def status_style(status):
return f"color: {RED}; font-weight: 600" if status >= 500 else ""
def history_items(alerts):
# The number of alerts varies, so this is the one place that builds components; the Column holding it persists.
if not alerts:
return [pn.ui.Typography("No alerts have fired yet. Pick an incident profile and go live.", sx={"color": "text.secondary", "p": 2})]
items = []
for r in alerts:
rule = r["rule"]
resolved = r["resolved"] is not None
ended = f"{r['resolved']:%H:%M:%S}" if resolved else "ongoing"
duration = f"{(r['resolved'] - r['started']).seconds} s" if resolved else "active"
peak = f"{rule['fmt'].format(r['peak'])} {rule['unit']}" if resolved else "pending"
body = (
f"**Threshold** {rule['threshold']} {rule['unit']} · **Peak** {peak} · "
f"**Window** {r['started']:%H:%M:%S} to {ended} · **Severity** {rule['severity']}"
)
status = "✓" if resolved else "●"
items.append((f"{status} {rule['name']} · {r['started']:%H:%M:%S} · {duration}", pn.ui.Markdown(body)))
return [pn.ui.Accordion(*items, active=[0], title_variant="subtitle1", margin=0)]
def about_text(seed, interval):
return f"""
All telemetry is generated in this session by a seeded pseudo-random simulator; nothing is collected or sent anywhere.
| Setting | Value |
|---|---|
| Seed | `{seed}` |
| Update interval | {interval} ms per simulated second (change in the sidebar) |
| History kept | {WINDOW} ticks ({WINDOW // 60} simulated minutes), older samples roll off |
| Capacity model | {REPLICA_RPS} req/s per replica, latency grows quadratically above 70% saturation |
| Alert rules | {", ".join(f"{r['name']} (> {r['threshold']} {r['unit']})" for r in RULES)} |
| Hysteresis | fire after {FIRE_AFTER} consecutive breaches, resolve after {RESOLVE_AFTER} clear ticks |
Try **Traffic surge**, watch saturation and latency climb, then drag **Replicas** up to 8 to recover.
"""
Building blocks#
Charts are Bokeh figures backed by a ColumnDataSource, so ticks stream single rows into them instead of re-rendering the plot.
def chart_card(title, y_label, lines, threshold=None, band=None):
"""Return a titled Paper holding a streaming time-series chart, and the source that feeds it."""
source = ColumnDataSource(data={})
fig = figure(
height=250, x_axis_type="datetime", toolbar_location=None, y_axis_label=y_label,
sizing_mode="stretch_width", min_border_left=50,
)
fig.xaxis.formatter = DatetimeTickFormatter(minutes="%H:%M", seconds="%H:%M:%S")
fig.y_range.start = 0
if band:
fig.varea(x="time", y1=band[0], y2=band[1], source=source, color=lines[-1][2], alpha=0.12)
renderers = [
fig.line(x="time", y=field, source=source, color=color, line_width=2, legend_label=label)
for field, label, color in lines
]
if threshold is not None:
fig.add_layout(Span(location=threshold, dimension="width", line_color=RED, line_dash="dashed", line_width=1.5))
fig.add_tools(HoverTool(
renderers=renderers[-1:], mode="vline",
tooltips=[("Time", "@time{%H:%M:%S}"), *((label, f"@{field}{{0,0.[000]}}") for field, label, _ in lines)],
formatters={"@time": "datetime"},
))
legend = fig.legend[0]
legend.orientation, legend.background_fill_alpha, legend.border_line_alpha = "horizontal", 0, 0
legend.visible = len(lines) > 1
fig.add_layout(legend, "above")
fig.grid.grid_line_alpha = 0.4
card = pn.ui.Paper(
pn.ui.Typography(title, variant="subtitle1", sx={"fontWeight": 600, "px": 1}),
pn.ui.Bokeh(fig, height=250), sx={"p": 1.5},
)
return card, source
def trend(label, color, plot_type, higher_is_worse=True):
return pn.ui.Trend(
label=label, plot_type=plot_type, plot_color=color, height=150, sizing_mode="stretch_width",
pos_color=RED if higher_is_worse else TEAL, neg_color=TEAL if higher_is_worse else RED,
)
def kpi_cell(obj, size):
return pn.ui.Grid(pn.ui.Paper(obj, sx={"p": 1, "height": "100%"}, margin=0, height=170), size=size)
Dashboard#
The monitor holds all state as parameters. Every pane, tile, gauge and the payload dialog is built once in __init__ and bound to those parameters. The periodic callback is created once, paused, and toggled through its running parameter. Charts, trend indicators and the request table are the exception: they stream into bounded buffers imperatively. Alerts use hysteresis and notify only on state transitions.
class ModelServingMonitor(pn.viewable.Viewer):
"""Live view of request rate, latency, errors and drift for a simulated model endpoint."""
running = param.Boolean(default=False, doc="Whether the simulation advances.")
profile = param.Selector(default="Steady", objects=list(PROFILES), doc="Traffic profile that drives the simulation.")
replicas = param.Integer(default=4, bounds=(1, 10), doc="Serving replicas; each absorbs roughly 90 req/s.")
interval = param.Selector(default=500, objects={"Fast (250 ms)": 250, "Normal (500 ms)": 500, "Slow (1 s)": 1000}, doc="Update interval.")
seed = param.Integer(default=SEED, doc="Random seed used on reset.")
reset = param.Event(doc="Restart the simulation from the seed.")
health = param.Selector(default="Healthy", objects=["Healthy", "Degraded", "Critical"])
saturation = param.Number(default=0, bounds=(0, 100), doc="Latest replica saturation in percent.")
availability = param.Number(default=100, bounds=(0, 100), doc="Request-weighted availability over the last two minutes.")
active_alerts = param.Dict(default={}, doc="Firing alert records keyed by rule.")
alerts = param.List(default=[], doc="Alert records, newest first.")
inspected = param.Dict(default={}, doc="Payload of the request shown in the dialog.")
dialog_open = param.Boolean(default=False, doc="Whether the payload dialog is visible.")
def __init__(self, **params):
super().__init__(**params)
self._callback = pn.state.add_periodic_callback(self._tick, period=self.interval, start=False)
self._build()
self._restart()
# Simulation lifecycle
def _restart(self):
self._sim = ServingSimulator(self.seed)
self._metrics = deque(maxlen=WINDOW)
self._payloads = {}
self._streaks = {rule["key"]: 0 for rule in RULES}
self._peaks = {}
# Pre-roll five minutes of steady traffic so the paused dashboard has context instead of empty charts.
rows = []
for _ in range(300):
rows += self._record(*self._sim.step("Steady", self.replicas))
frame = pd.DataFrame(self._metrics)
for source in self._sources.values():
source.data = {column: frame[column].to_numpy() for column in frame.columns}
tail = frame.tail(TREND_WINDOW)
for key, indicator in self._trends.items():
indicator.data = {"x": list(range(self._sim.tick - len(tail) + 1, self._sim.tick + 1)), "y": tail[key].round(TREND_DIGITS[key]).tolist()}
self._set_requests(pd.DataFrame(rows[-TABLE_ROWS:]))
self._update_kpis()
self.param.update(alerts=[], active_alerts={}, health="Healthy", inspected={}, dialog_open=False)
def _record(self, metrics, requests):
self._metrics.append(metrics)
for row, payload in requests:
self._payloads[row["request_id"]] = payload
return [row for row, _ in requests]
def _set_requests(self, frame):
self._table.value = frame.reset_index(drop=True)
self._payloads = {rid: self._payloads[rid] for rid in frame.request_id}
def _append_requests(self, rows):
# Clicks report a row position, so rows are appended rather than replaced to keep positions stable between ticks.
new = pd.DataFrame(rows)
if len(self._table.value) + len(new) > 2 * TABLE_ROWS:
self._set_requests(pd.concat([self._table.value.tail(TABLE_ROWS), new]))
else:
self._table.stream(new, follow=True)
def _tick(self):
rows = self._record(*self._sim.step(self.profile, self.replicas))
metrics = self._metrics[-1]
with pn.io.hold():
for source in self._sources.values():
source.stream({k: [v] for k, v in metrics.items()}, rollover=WINDOW)
x = self._sim.tick
for key, indicator in self._trends.items():
indicator.stream({"x": [x], "y": [round(metrics[key], TREND_DIGITS[key])]}, rollover=TREND_WINDOW)
self._update_kpis()
self._append_requests(rows)
notices = self._evaluate_alerts(metrics)
# Notifications sent inside pn.io.hold() currently fail on the client with "reference isn't known".
if pn.state.notifications:
for severity, message in notices:
getattr(pn.state.notifications, severity)(message, duration=5000)
def _update_kpis(self):
self.param.update(
saturation=round(float(self._metrics[-1]["saturation"])),
availability=availability(pd.DataFrame(self._metrics).tail(120)),
)
def _evaluate_alerts(self, metrics):
# Records are replaced rather than mutated so that every change is a new parameter value the panes can react to.
active, alerts = dict(self.active_alerts), list(self.alerts)
notices, resolved = [], []
for rule in RULES:
key, value = rule["key"], float(metrics[rule["key"]])
record = active.get(key)
breached = value > rule["threshold"]
self._streaks[key] = self._streaks[key] + 1 if breached == (record is None) else 0
if record is not None:
self._peaks[key] = max(self._peaks[key], value)
if record is None and self._streaks[key] >= FIRE_AFTER:
record = dict(rule=rule, started=metrics["time"], resolved=None, peak=value)
active[key], self._peaks[key], self._streaks[key] = record, value, 0
alerts.insert(0, record)
notices.append((rule["severity"], f"{rule['name']}: {rule['fmt'].format(value)} {rule['unit']}"))
elif record is not None and self._streaks[key] >= RESOLVE_AFTER:
done = {**record, "resolved": metrics["time"], "peak": self._peaks.pop(key)}
alerts = [done if r is record else r for r in alerts]
del active[key]
self._streaks[key] = 0
resolved.append(rule["name"])
if notices or resolved:
self.param.update(active_alerts=active, alerts=alerts, health=health_for(active))
if resolved:
notices.append(("success", f"Resolved: {', '.join(resolved)}"))
return notices
@param.depends("running", watch=True)
def _toggle_running(self):
self._callback.running = self.running
@param.depends("interval", watch=True)
def _update_interval(self):
self._callback.period = self.interval
@param.depends("reset", watch=True)
def _reset(self):
self.param.update(running=False, profile="Steady", replicas=4)
self._restart()
if pn.state.notifications:
pn.state.notifications.info(f"Simulation reset with seed {self.seed}.")
def _inspect(self, event):
row = self._table.value.iloc[event.row]
self.param.update(inspected=self._payloads.get(row.request_id, {}), dialog_open=True)
def _sync_dialog_open(self, event):
# Binding `open` only flows param -> dialog; mirror client-side closes back so the next click reopens it.
self.dialog_open = event.new
# Views
def _build(self):
self._signals = [
pn.ui.Alert(
pn.bind(signal_text, self.param.active_alerts, rule["key"]), title=rule["title"],
severity=pn.bind(signal_severity, self.param.active_alerts, rule),
variant=pn.bind(signal_variant, self.param.active_alerts, rule["key"]), margin=0, min_height=88,
)
for rule in RULES
]
gauge = pn.ui.Gauge(
value=self.param.saturation, label="Saturation", format="{value}%", bounds=(0, 100), height=170, sizing_mode="stretch_width",
colors=[(0.7, TEAL), (0.9, AMBER), (1, RED)], annulus_width=8, title_size=12,
custom_opts={"pointer": {"width": 4}, "detail": {"fontSize": 20, "offsetCenter": [0, "60%"]}, "axisLabel": {"show": False}, "splitLine": {"length": 6}},
)
number = pn.ui.Number(
value=self.param.availability, label="Availability (2 min)", format="{value:.2f}%", font_size="34pt", title_size="11pt",
colors=[(99.0, RED), (99.5, AMBER), (100, TEAL)], sizing_mode="stretch_width", align="center",
)
self._trends = {
"rps": trend("Throughput (req/s)", INDIGO, "area", higher_is_worse=False),
"p95": trend("p95 latency (ms)", AMBER, "line"),
"error_rate": trend("Error rate (%)", RED, "bar"),
"drift": trend("Feature drift (PSI)", TEAL, "step"),
}
kpi_size = {"xs": 12, "sm": 6, "md": 4, "xl": 2}
self._kpis = pn.ui.Grid(
*(kpi_cell(indicator, kpi_size) for indicator in self._trends.values()),
kpi_cell(gauge, kpi_size),
kpi_cell(pn.ui.Column(
pn.ui.VSpacer(), number,
pn.ui.Typography("SLO target 99.5%", variant="caption", align="center", sx={"color": "text.secondary"}), pn.ui.VSpacer(),
), kpi_size),
container=True, spacing=2,
)
charts = {
"Latency": chart_card("Latency", "ms", [("p50", "p50", INDIGO), ("p95", "p95", AMBER)], threshold=250, band=("p50", "p95")),
"Throughput": chart_card("Throughput", "req/s", [("rps", "requests / s", INDIGO)], band=(0, "rps")),
"Error rate": chart_card("Error rate", "%", [("error_rate", "5xx share", RED)], threshold=2.0, band=(0, "error_rate")),
"Feature drift": chart_card("Feature drift", "PSI", [("drift", "population stability index", TEAL)], threshold=0.2),
}
self._sources = {title: source for title, (_, source) in charts.items()}
self._charts = pn.ui.Grid(
*(pn.ui.Grid(card, size={"xs": 12, "lg": 6}) for card, _ in charts.values()),
container=True, spacing=2,
)
self._table = pn.ui.Tabulator(
pd.DataFrame(), show_index=False, height=360, layout="fit_columns", disabled=True, selectable=False,
titles={
"time": "Time", "request_id": "Request", "endpoint": "Endpoint", "status": "Status",
"latency_ms": "Latency (ms)", "score": "Score", "decision": "Decision",
},
formatters={"latency_ms": {"type": "progress", "min": 0, "max": 1500, "legend": True, "color": [TEAL, AMBER, RED]}},
stylesheets=[":host .tabulator-row { cursor: pointer; }"],
)
self._table.style.map(status_style, subset=["status"])
self._table.on_click(self._inspect)
self._dialog = pn.ui.Dialog(
pn.ui.Alert(
pn.bind(payload_summary, self.param.inspected), severity=pn.bind(payload_severity, self.param.inspected),
margin=(0, 0, 10, 0),
),
pn.ui.CodeEditor(
value=pn.bind(payload_json, self.param.inspected), visible=self.param.inspected.rx.bool(),
language="json", readonly=True, height=300,
),
open=self.param.dialog_open, title="Request payload", show_close_button=True,
close_on_click=True, width_option="sm",
)
self._dialog.param.watch(self._sync_dialog_open, "open")
self._page = pn.ui.Page(
title="Model Serving Monitor",
header=[
pn.ui.Chip(
label="fraud-scorer 3.2.1 · simulated", icon="memory", variant="outlined", color="default", size="small",
margin=(0, 10), sx={"color": "inherit", "borderColor": "rgba(255,255,255,0.5)"},
),
pn.ui.Chip(
label=self.param.health, icon=pn.bind(HEALTH_ICONS.get, self.param.health),
color=pn.bind(HEALTH_COLORS.get, self.param.health), size="small", margin=(0, 10),
),
],
sidebar=[self._sidebar()],
main=[self._main()],
main_width={"xs": "100%", "xl": 1680},
theme_config={
"light": {"palette": {"primary": {"main": INDIGO}, "background": {"default": "#F6F7FB"}}, "shape": {"borderRadius": 10}},
"dark": {"palette": {"primary": {"main": "#818CF8"}, "background": {"default": "#0B1020", "paper": "#131A2E"}}, "shape": {"borderRadius": 10}},
},
)
def _sidebar(self):
running = self.param.running.rx
return pn.ui.Column(
pn.ui.Row(
pn.ui.Toggle.from_param(
self.param.running, label=running.where("Pause", "Go live"), icon=running.where("pause", "play_arrow"),
color="primary", description="", sizing_mode="stretch_width",
),
pn.ui.Button.from_param(self.param.reset, label="Reset", icon="restart_alt", variant="outlined", sizing_mode="stretch_width"),
),
pn.ui.Divider(margin=(10, 10)),
pn.ui.Typography("Traffic profile", variant="overline", margin=(0, 10)),
pn.ui.RadioBoxGroup.from_param(self.param.profile, label="", sizing_mode="stretch_width"),
pn.ui.Alert(
pn.bind(profile_blurb, self.param.profile), title=self.param.profile,
severity=pn.bind(profile_severity, self.param.profile), margin=(0, 10),
),
pn.ui.Divider(margin=(10, 10)),
pn.ui.Typography("Capacity", variant="overline", margin=(0, 10)),
pn.ui.IntSlider.from_param(self.param.replicas, label="Replicas", sizing_mode="stretch_width", marks=True),
pn.ui.Typography(
pn.bind(capacity_text, self.param.replicas), variant="caption", margin=(0, 10), sx={"color": "text.secondary"},
),
pn.ui.Select.from_param(self.param.interval, label="Update interval", sizing_mode="stretch_width"),
)
def _main(self):
tabs = pn.ui.Tabs(
("Recent requests", pn.ui.Column(
pn.ui.Typography("Click a row to inspect the request payload.", variant="caption", sx={"color": "text.secondary"}, margin=(10, 0)),
self._table,
)),
("Alert history", pn.ui.Column(objects=pn.bind(history_items, self.param.alerts), height=400, scroll="y-auto")),
("About", pn.ui.Markdown(pn.bind(about_text, self.param.seed, self.param.interval), margin=(10, 16))),
margin=0,
)
return pn.ui.Column(
pn.ui.Grid(*(pn.ui.Grid(tile, size={"xs": 12, "sm": 6, "lg": 3}) for tile in self._signals), container=True, spacing=2),
self._kpis,
self._charts,
pn.ui.Paper(tabs, sx={"p": 1.5}, margin=0),
self._dialog,
styles={"gap": "16px"}, margin=(10, 0),
)
def __panel__(self):
# The Page is built once in __init__ so the preview and the served app share one set of linked components.
return self._page
Preview#
The same monitor instance is displayed inline here and served below.
app = ModelServingMonitor()
app
if pn.state.served:
app.servable()
