#!/usr/bin/env python3
"""Collect Perforce topology data and render a Mermaid diagram or other output formats.
The script can either:
1. run the live commands needed to collect topology data (p4 servers/p4 topology), or
2. render from a previously collected JSON file.
Outputs:
- JSON intermediate file with raw command output and normalized records
- Mermaid diagram source
- Markdown file wrapping the Mermaid diagram in a fenced block
- AsciiDoc block wrapping the Mermaid diagram for direct inclusion
- SVG diagram with drag-and-save support
- HTML page with interactive D3.js diagram
If you have Graphviz installed, you can use the --layout option to generate a more compact SVG layout.
If you have the fast-sugiyama python pip package installed, it will be used first
for a better layout unless --layout is set to one of its various options.
If neither fast-sugiyama nor Graphviz are available, a simple grid layout will be used.
"""
from __future__ import annotations
import argparse
import json
import ipaddress
import re
import shlex
import subprocess
import sys
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable
from xml.sax.saxutils import escape
IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
SERVER_LINE_RE = re.compile(r"^(?P<serverid>\S+)\s+(?P<service>server|proxy)\s+(?P<rest>.*)$")
GRAPHVIZ_PROGRAMS = ("dot", "neato", "fdp", "sfdp", "circo", "twopi", "osage")
GRAPHVIZ_FORCE_DIRECTED = {"neato", "fdp", "sfdp", "circo", "twopi"}
LAYOUT_CHOICES = ("auto", "sugiyama", *GRAPHVIZ_PROGRAMS, "grid")
@dataclass
class CommandCapture:
command: str
output: object
@dataclass
class ServerRecord:
serverid: str
service: str
server_type: str
display_name: str | None
address: str | None
port: str | None
host: str | None
ip: str | None
version: str | None
target: str | None
target_source: str | None
host_target: str | None
participates_in_topology: bool
notes: list[str]
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--input-json",
type=Path,
help="Use an existing JSON intermediate file instead of running live commands.",
)
parser.add_argument(
"--json-output",
type=Path,
default=Path("p4_topology_data.json"),
help="Path to write the JSON intermediate file.",
)
parser.add_argument(
"--mermaid-output",
type=Path,
default=Path("p4_topology.mmd"),
help="Path to write the Mermaid diagram source.",
)
parser.add_argument(
"--markdown-output",
type=Path,
default=Path("p4_topology.md"),
help="Path to write the Markdown file containing the Mermaid diagram.",
)
parser.add_argument(
"--asciidoc-output",
type=Path,
default=Path("p4_topology.adoc"),
help="Path to write an AsciiDoc Mermaid block.",
)
parser.add_argument(
"--html-output",
type=Path,
default=Path("p4_topology.html"),
help="Path to write the interactive D3.js HTML diagram.",
)
parser.add_argument(
"--svg-output",
type=Path,
default=Path("p4_topology.svg"),
help="Path to write a laid-out SVG diagram with drag-and-save support.",
)
parser.add_argument(
"--layout",
choices=LAYOUT_CHOICES,
default="auto",
help="SVG layout engine (auto tries fast-sugiyama, then graphviz dot, then a simple grid).",
)
parser.add_argument(
"--title",
default="P4 server topology",
help="Title to embed in the generated Markdown output.",
)
parser.add_argument(
"--p4port",
help="Perforce server port to use for live data collection.",
)
parser.add_argument(
"--p4user",
help="Perforce user to use for live data collection.",
)
return parser.parse_args()
def run_command(command: list[str]) -> str:
try:
result = subprocess.run(command, check=True, capture_output=True, text=True)
except FileNotFoundError as exc:
raise RuntimeError(f"Command not found: {command[0]}") from exc
except subprocess.CalledProcessError as exc:
stderr = exc.stderr.strip()
detail = f"\n{stderr}" if stderr else ""
raise RuntimeError(f"Command failed: {' '.join(command)}{detail}") from exc
return result.stdout.strip()
def p4_value(record: dict, *names: str) -> str | None:
for name in names:
value = record.get(name)
if value is not None and str(value).strip():
return str(value).strip()
return None
def parse_server_address(address: str | None) -> str | None:
if not address:
return None
value = address
if value.startswith("ssl:"):
value = value[4:]
if ":" not in value:
return None
host, _port = value.rsplit(":", 1)
if not host or host.isdigit():
return None
return host
def parse_server_port(address: str | None) -> str | None:
if not address or ":" not in address:
return None
return address.rsplit(":", 1)[1]
def is_loopback_address(address: str | None) -> bool:
host = parse_server_address(address)
if not host:
return False
try:
return ipaddress.ip_address(host).is_loopback
except ValueError:
return False
def parse_p4_servers(output: str) -> list[ServerRecord]:
records: list[ServerRecord] = []
for raw_line in output.splitlines():
line = raw_line.strip()
if not line:
continue
match = SERVER_LINE_RE.match(line)
if not match:
continue
serverid = match.group("serverid")
service = match.group("service")
before_quote = match.group("rest").split("'", 1)[0].strip()
parts = before_quote.split()
if len(parts) >= 2:
if len(parts) == 2:
address = None
server_type = parts[1]
else:
address = parts[0]
server_type = parts[1]
else:
address = None
server_type = "unknown"
host = parse_server_address(address)
records.append(
ServerRecord(
serverid=serverid,
service=service,
server_type=server_type,
display_name=None,
address=address,
port=parse_server_port(address),
host=host,
ip=None,
version=None,
target=None,
target_source=None,
host_target=None,
participates_in_topology=False,
notes=[],
)
)
return records
def parse_p4_server_records(output: list[dict]) -> list[ServerRecord]:
records: list[ServerRecord] = []
for record in output:
services = p4_value(record, "services", "Services", "service", "Service") or "unknown"
if services == "git-connector":
continue
serverid = p4_value(record, "serverid", "serverID", "ServerID", "name", "Name")
if not serverid:
continue
address = p4_value(record, "address", "Address", "externalAddress", "ExternalAddress")
service = p4_value(record, "type", "Type") or "server"
records.append(
ServerRecord(
serverid=serverid,
service=service,
server_type=services,
display_name=None,
address=address,
port=parse_server_port(address),
host=parse_server_address(address),
ip=None,
version=None,
target=None,
target_source=None,
host_target=None,
participates_in_topology=False,
notes=[],
)
)
return records
def parse_nslookup_ip(output: str) -> str | None:
name_seen = False
candidate: str | None = None
for raw_line in output.splitlines():
line = raw_line.strip()
if line.startswith("Name:"):
name_seen = True
continue
if line.startswith("Address:"):
value = line.split(":", 1)[1].strip()
if "#" in value:
continue
if IPV4_RE.fullmatch(value):
if name_seen:
return value
candidate = value
return candidate
def parse_topology(output: str | list[dict], known_serverids: set[str]) -> dict[str, dict[str, str | bool | None]]:
if isinstance(output, list):
topology: dict[str, dict[str, str | bool]] = {}
for record in output:
subject = p4_value(record, "serverid", "serverID", "ServerID", "server")
target = p4_value(
record,
"parent",
"parentServer",
"parentServerID",
"TargetServerID",
)
if subject in known_serverids:
topology[subject] = {
"target": target,
"version": p4_value(record, "version", "Version"),
"seen": True,
}
return topology
topology: dict[str, dict[str, str | bool]] = {}
for raw_line in output.splitlines():
line = raw_line.strip()
if not line or line.startswith("Topology command was run on server"):
continue
tokens = line.split()
known_tokens = [token for token in tokens if token in known_serverids]
if not known_tokens:
continue
subject = known_tokens[0]
target = known_tokens[-1] if len(known_tokens) > 1 else None
topology[subject] = {
"target": target,
"version": None,
"seen": True,
}
return topology
def add_pseudo_topology_servers(records: list[ServerRecord], topology_output: list[dict] | str) -> None:
if not isinstance(topology_output, list):
return
known_serverids = {record.serverid for record in records}
selected_records: dict[tuple[str, str], dict] = {}
for topology_record in topology_output:
server_type = p4_value(topology_record, "type", "Type")
if server_type not in {"broker", "proxy"}:
continue
address = p4_value(topology_record, "serverAddress", "ServerAddress")
target_serverid = p4_value(topology_record, "targetServerID", "TargetServerID")
if not address or not target_serverid or target_serverid not in known_serverids:
continue
key = (server_type, target_serverid)
current = selected_records.get(key)
if current is not None and (
is_loopback_address(address)
or not is_loopback_address(p4_value(current, "serverAddress", "ServerAddress"))
):
continue
selected_records[key] = topology_record
for (server_type, target_serverid), topology_record in selected_records.items():
address = p4_value(topology_record, "serverAddress", "ServerAddress")
if not address:
continue
node_address = parse_server_address(address) or address
safe_address = re.sub(r"[^A-Za-z0-9]+", "-", node_address).strip("-")
pseudo_id = f"{server_type}-{safe_address}"
if pseudo_id in known_serverids:
continue
records.append(
ServerRecord(
serverid=pseudo_id,
service="pseudo",
server_type=server_type,
display_name=address,
address=address,
port=parse_server_port(address),
host=parse_server_address(address),
ip=parse_server_address(address) if not is_loopback_address(address) else None,
version=p4_value(topology_record, "version", "Version"),
target=target_serverid,
target_source="topology",
host_target=(
target_serverid
if parse_server_address(p4_value(topology_record, "targetAddress", "TargetAddress"))
== "127.0.0.1"
else None
),
participates_in_topology=True,
notes=[f"{server_type} for {target_serverid}"],
)
)
def topology_own_addresses(topology_output: list[dict] | str) -> dict[str, str]:
"""Map serverid -> its own listening address, from topology records where ServerID is the record's own id."""
addresses: dict[str, str] = {}
if not isinstance(topology_output, list):
return addresses
for record in topology_output:
serverid = p4_value(record, "serverid", "serverID", "ServerID")
address = p4_value(record, "serverAddress", "ServerAddress")
if not serverid or not address or is_loopback_address(address):
continue
if serverid not in addresses:
addresses[serverid] = address
return addresses
def enrich_server_records(records: list[ServerRecord], topology_output: list[dict] | str) -> None:
topology_map = parse_topology(topology_output, {record.serverid for record in records})
own_addresses = topology_own_addresses(topology_output)
master_host = next(
(record.host for record in records if record.server_type == "commit-server"),
None,
)
for record in records:
if not record.host:
own_address = own_addresses.get(record.serverid)
if own_address:
record.address = record.address or own_address
record.host = parse_server_address(own_address)
record.port = record.port or parse_server_port(own_address)
if record.host and not is_loopback_address(own_address):
record.ip = record.ip or record.host
master_host = master_host or (
record.host if record.server_type == "commit-server" else master_host
)
topo_entry = topology_map.get(record.serverid)
if topo_entry and isinstance(topo_entry.get("version"), str):
record.version = topo_entry["version"]
if record.server_type == "commit-server":
record.target = None
record.target_source = "root"
continue
if topo_entry:
record.target = topo_entry["target"] if isinstance(topo_entry["target"], str) else None
record.target_source = "topology" if record.target else None
if isinstance(topo_entry.get("version"), str):
record.version = topo_entry["version"]
record.participates_in_topology = True
elif record.server_type == "forwarding-standby" and record.host and record.host == master_host:
record.target = "master"
record.target_source = "inferred-from-address"
record.notes.append("Configured, but not present in captured topology output")
def collect_live_data(p4port: str | None = None, p4user: str | None = None) -> dict:
try:
from P4 import P4, P4Exception
except ModuleNotFoundError as exc:
raise RuntimeError("p4python is required for live data collection; install requirements.txt") from exc
p4_options: list[str] = []
if p4port:
p4_options.extend(["-p", p4port])
if p4user:
p4_options.extend(["-u", p4user])
p4 = P4()
if p4port:
p4.port = p4port
if p4user:
p4.user = p4user
p4_servers_command = ["p4", *p4_options, "servers"]
topology_command = ["p4", *p4_options, "topology", "-t", "2"]
try:
p4.connect()
p4_servers_output = p4.run("servers")
records = parse_p4_server_records(p4_servers_output)
topology_output = p4.run("topology", "-t", "2")
except P4Exception as exc:
raise RuntimeError(f"Perforce command failed: {exc}") from exc
finally:
if p4.connected():
p4.disconnect()
add_pseudo_topology_servers(records, topology_output)
nslookup_results: dict[str, CommandCapture] = {}
hosts = sorted({record.host for record in records if record.host})
for host in hosts:
try:
output = run_command(["nslookup", host])
except RuntimeError as exc:
output = str(exc)
nslookup_results[host] = CommandCapture(
command=f"nslookup {host}",
output=output,
)
ip_by_host = {
host: parse_nslookup_ip(capture.output)
for host, capture in nslookup_results.items()
}
for record in records:
if record.host:
record.ip = ip_by_host.get(record.host)
enrich_server_records(records, topology_output)
return {
"collected_at": datetime.now(timezone.utc).isoformat(),
"commands": {
"p4_servers": asdict(
CommandCapture(command=" ".join(p4_servers_command), output=json.dumps(p4_servers_output))
),
"p4_topology": asdict(
CommandCapture(command=" ".join(topology_command), output=topology_output)
),
"nslookup": {host: asdict(capture) for host, capture in nslookup_results.items()},
},
"servers": [asdict(record) for record in records],
}
def load_data(path: Path) -> dict:
with path.open("r", encoding="utf-8") as handle:
data = json.load(handle)
commands = data.get("commands", {})
topology_capture = commands.get("p4_topology", {})
p4_topology = topology_capture.get("output")
if isinstance(p4_topology, str):
try:
topology_capture["output"] = json.loads(p4_topology)
except json.JSONDecodeError:
pass
if data.get("servers"):
topology_output = topology_capture.get("output")
if isinstance(topology_output, list):
records = [
ServerRecord(
serverid=server["serverid"],
service=server.get("service", "server"),
server_type=server.get("server_type", "unknown"),
display_name=server.get("display_name"),
address=server.get("address"),
port=server.get("port") or parse_server_port(server.get("address")),
host=server.get("host"),
ip=server.get("ip"),
version=server.get("version"),
target=server.get("target", server.get("parent")),
target_source=server.get("target_source", server.get("parent_source")),
host_target=server.get("host_target"),
participates_in_topology=server.get("participates_in_topology", False),
notes=server.get("notes", []),
)
for server in data["servers"]
if server.get("service", "server") != "pseudo"
]
add_pseudo_topology_servers(records, topology_output)
enrich_server_records(records, topology_output)
data["servers"] = [asdict(record) for record in records]
return data
p4_servers = commands.get("p4_servers", {}).get("output")
p4_topology = topology_capture.get("output")
if not p4_servers or not p4_topology:
return data
try:
server_output = json.loads(p4_servers) if isinstance(p4_servers, str) else p4_servers
topology_output = p4_topology
records = parse_p4_server_records(server_output)
except (TypeError, json.JSONDecodeError):
return data
enrich_server_records(records, topology_output)
data["servers"] = [asdict(record) for record in records]
return data
def save_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2, sort_keys=True)
handle.write("\n")
def p4server_label_lines(server: dict) -> list[str]:
fields = [
f"p4server: {server['serverid']}",
f"type: {server['server_type']}",
f"port: {server.get('port') or parse_server_port(server.get('address')) or '-'}",
]
if server.get("target"):
fields.append(f"target: {server['target']}")
if server.get("version"):
fields.append(f"version: {server['version']}")
for note in server.get("notes", []):
fields.append(f"note: {note}")
return fields
def host_label_lines(server: dict) -> list[str]:
return [
"host server",
f"address: {server.get('host') or 'unknown'}",
f"ip: {server.get('ip') or 'unknown'}",
]
def format_p4server_label(server: dict) -> str:
return "<br/>".join(p4server_label_lines(server))
def format_host_label(server: dict) -> str:
return "<br/>".join(host_label_lines(server))
def node_class(server_type: str) -> str:
normalized_type = server_type.lower()
if normalized_type in {"commit-server", "standard"}:
return "commit"
if normalized_type == "edge-server":
return "edge"
if normalized_type in {"forwarding-standby", "standby", "forwarding-replica", "replica"}:
return "replica"
if normalized_type == "proxy":
return "proxy"
if normalized_type == "broker":
return "broker"
return "commit"
def iter_diagram_servers(servers: list[dict]) -> Iterable[dict]:
by_id = {server["serverid"]: server for server in servers}
yielded: set[str] = set()
roots = [
server
for server in servers
if server["serverid"] == "master" or server["server_type"] == "commit-server"
]
for server in sorted(roots, key=lambda item: item["serverid"]):
yielded.add(server["serverid"])
yield server
for server in sorted(servers, key=lambda item: item["serverid"]):
if server["serverid"] in yielded:
continue
if server.get("target") or server["serverid"] == "master":
yielded.add(server["serverid"])
yield server
def render_mermaid(data: dict) -> str:
servers = data["servers"]
diagram_servers = list(iter_diagram_servers(servers))
lines = ["graph TD"]
for server in diagram_servers:
serverid = server["serverid"]
lines.append(f' {serverid}["{format_p4server_label(server)}"]')
if not server.get("host_target"):
host_node = f"host-{serverid}"
lines.append(f' {host_node}["{format_host_label(server)}"]')
lines.append("")
for server in diagram_servers:
serverid = server["serverid"]
host_node = f"host-{server.get('host_target') or serverid}"
lines.append(f" {serverid} --> {host_node}")
target = server.get("target")
if target:
lines.append(f" {serverid} --> {target}")
lines.extend(
[
"",
" classDef commit fill:#d7f0ff,stroke:#1f5f8b,stroke-width:2px,color:#111;",
" classDef edge fill:#dff7df,stroke:#2f6b2f,stroke-width:2px,color:#111;",
" classDef replica fill:#fff1cc,stroke:#8a6d1d,stroke-width:2px,color:#111;",
" classDef proxy fill:#f4ddff,stroke:#7b3f98,stroke-width:2px,color:#111;",
" classDef broker fill:#ffe0cc,stroke:#a64b00,stroke-width:2px,color:#111;",
" classDef host fill:#fff,stroke:#666,stroke-width:1px,color:#111;",
"",
]
)
classes: dict[str, list[str]] = {"commit": [], "edge": [], "replica": [], "proxy": [], "broker": []}
for server in diagram_servers:
classes[node_class(server["server_type"])].append(server["serverid"])
if not server.get("host_target"):
classes.setdefault("host", []).append(f"host-{server['serverid']}")
for class_name, members in classes.items():
if members:
lines.append(f" class {','.join(members)} {class_name};")
return "\n".join(lines) + "\n"
def render_markdown(title: str, mermaid: str, data: dict) -> str:
unresolved = [
server["serverid"]
for server in sorted(data["servers"], key=lambda item: item["serverid"])
if server["serverid"] != "master"
and server["server_type"] != "commit-server"
and not server.get("target")
and not server.get("participates_in_topology")
]
lines = [f"# {title}", "", "```mermaid", mermaid.rstrip(), "```"]
if unresolved:
lines.extend(
[
"",
"Unplaced configured entries:",
"",
*[f"- {serverid}" for serverid in unresolved],
]
)
return "\n".join(lines) + "\n"
def render_asciidoc(title: str, mermaid: str, data: dict) -> str:
unresolved = [
server["serverid"]
for server in sorted(data["servers"], key=lambda item: item["serverid"])
if server["serverid"] != "master"
and server["server_type"] != "commit-server"
and not server.get("target")
and not server.get("participates_in_topology")
]
target = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "p4-server-topology"
lines = [
f". {title}",
f"[mermaid, target=\"{target}\"]",
"----",
mermaid.rstrip(),
"----",
]
if unresolved:
lines.extend(
[
"",
"Unplaced configured entries:",
"",
*[f"* {serverid}" for serverid in unresolved],
]
)
return "\n".join(lines) + "\n"
def write_text(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:
handle.write(content)
def render_html(title: str, data: dict) -> str:
diagram_servers = list(iter_diagram_servers(data["servers"]))
unresolved = [
server["serverid"]
for server in sorted(data["servers"], key=lambda item: item["serverid"])
if server["serverid"] != "master"
and server["server_type"] != "commit-server"
and not server.get("target")
and not server.get("participates_in_topology")
]
diagram_data = {"servers": diagram_servers, "unresolved": unresolved}
payload = json.dumps(diagram_data, separators=(",", ":")).replace("</", "<\\/")
title_json = json.dumps(title)
return """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>""" + title_json + """</title>
<script src="https://cdn.jsdelivr.net/npm/d3@7"></script>
<style>
:root { color-scheme: light; font-family: ui-sans-serif, system-ui, sans-serif; }
body { margin: 0; background: #f4f7f9; color: #17212b; }
header { padding: 18px 24px 12px; background: #17212b; color: white; }
h1 { margin: 0; font-size: 20px; font-weight: 650; }
p { margin: 6px 0 0; color: #b8c6d1; font-size: 13px; }
#diagram { height: calc(100vh - 88px); min-height: 520px; }
svg { display: block; width: 100%; height: 100%; cursor: grab; }
svg:active { cursor: grabbing; }
.link { fill: none; stroke: #778692; stroke-width: 1.5px; marker-end: url(#arrow); }
.link.host-link { stroke: #aab5bd; stroke-dasharray: 4 3; marker-end: none; }
.node rect { stroke-width: 2px; rx: 7px; }
.node text { pointer-events: none; font-size: 12px; }
.node .heading { font-weight: 700; font-size: 13px; }
.host rect { fill: #fff; stroke: #66737c; }
.unresolved { position: fixed; right: 18px; bottom: 16px; padding: 10px 12px; background: #fff8e6; border: 1px solid #d9ad4d; border-radius: 6px; font-size: 12px; }
</style>
</head>
<body>
<header><h1>""" + title_json + """</h1><p>Drag nodes to reposition them. Scroll to zoom, drag the background to pan.</p></header>
<div id="diagram"></div>
<script>
const data = """ + payload + """;
const colors = {
commit: { fill: '#d7f0ff', stroke: '#1f5f8b' },
replica: { fill: '#fff1cc', stroke: '#8a6d1d' },
proxy: { fill: '#f4ddff', stroke: '#7b3f98' },
broker: { fill: '#ffe0cc', stroke: '#a64b00' },
edge: { fill: '#dff7df', stroke: '#2f6b2f' }
};
const width = document.querySelector('#diagram').clientWidth;
const height = document.querySelector('#diagram').clientHeight;
const svg = d3.select('#diagram').append('svg').attr('viewBox', [0, 0, width, height]);
const viewport = svg.append('g');
const defs = svg.append('defs');
defs.append('marker').attr('id', 'arrow').attr('viewBox', '0 -5 10 10').attr('refX', 20)
.attr('refY', 0).attr('markerWidth', 6).attr('markerHeight', 6).attr('orient', 'auto')
.append('path').attr('d', 'M0,-5L10,0L0,5').attr('fill', '#778692');
const nodes = [];
const links = [];
const nodeById = new Map();
const category = type => {
const value = type.toLowerCase();
if (value === 'edge-server') return 'edge';
if (['forwarding-standby', 'standby', 'forwarding-replica', 'replica'].includes(value)) return 'replica';
if (value === 'proxy') return 'proxy';
if (value === 'broker') return 'broker';
return 'commit';
};
data.servers.forEach(server => {
const p4 = { id: server.serverid, kind: 'p4', label: `p4server: ${server.serverid}`, type: server.server_type, port: server.port || '-', version: server.version || '-', target: server.target || '-' };
nodes.push(p4); nodeById.set(p4.id, p4);
if (!server.host_target) {
const host = { id: `host-${server.serverid}`, kind: 'host', label: 'host server', address: server.host || 'unknown', ip: server.ip || 'unknown' };
nodes.push(host); nodeById.set(host.id, host);
}
});
data.servers.forEach(server => {
const hostId = `host-${server.host_target || server.serverid}`;
links.push({ source: server.serverid, target: hostId, hostLink: true });
if (server.target && nodeById.has(server.target)) links.push({ source: server.serverid, target: server.target });
});
const linkSelection = viewport.append('g').selectAll('line').data(links).join('line').attr('class', d => d.hostLink ? 'link host-link' : 'link');
const nodeSelection = viewport.append('g').selectAll('g').data(nodes).join('g').attr('class', d => `node ${d.kind}`);
nodeSelection.append('rect').each(function(d) { const selection = d3.select(this); if (d.kind === 'p4') { const c = colors[category(d.type)]; selection.attr('x', -150).attr('y', -50).attr('width', 300).attr('height', 100).attr('fill', c.fill).attr('stroke', c.stroke); } else { selection.attr('x', -105).attr('y', -27).attr('width', 210).attr('height', 54); } });
nodeSelection.append('text').attr('class', 'heading').attr('x', d => d.kind === 'p4' ? -138 : -94).attr('y', d => d.kind === 'p4' ? -28 : -8).text(d => d.label);
nodeSelection.filter(d => d.kind === 'p4').append('text').attr('x', -138).attr('y', -8).text(d => `type: ${d.type}`);
nodeSelection.filter(d => d.kind === 'p4').append('text').attr('x', -138).attr('y', 12).text(d => {
return `port: ${d.port}`;
});
nodeSelection.filter(d => d.kind === 'p4' && d.target !== '-').append('text').attr('x', -138).attr('y', 32).text(d => `target: ${d.target}`);
nodeSelection.filter(d => d.kind === 'p4' && d.version !== '-').append('text').attr('x', -138).attr('y', 52).text(d => `version: ${d.version}`);
nodeSelection.filter(d => d.kind === 'host').append('text').attr('x', -94).attr('y', 10).text(d => `address: ${d.address}`);
nodeSelection.filter(d => d.kind === 'host').append('text').attr('x', -94).attr('y', 25).text(d => `ip: ${d.ip}`);
const simulation = d3.forceSimulation(nodes).force('link', d3.forceLink(links).id(d => d.id).distance(180)).force('charge', d3.forceManyBody().strength(-420)).force('center', d3.forceCenter(width / 2, height / 2)).force('collide', d3.forceCollide(125));
simulation.on('tick', () => { linkSelection.attr('x1', d => d.source.x).attr('y1', d => d.source.y).attr('x2', d => d.target.x).attr('y2', d => d.target.y); nodeSelection.attr('transform', d => `translate(${d.x},${d.y})`); });
nodeSelection.call(d3.drag().on('start', (event, d) => { if (!event.active) simulation.alphaTarget(0.3).restart(); d.fx = d.x; d.fy = d.y; }).on('drag', (event, d) => { d.fx = event.x; d.fy = event.y; }).on('end', (event, d) => { if (!event.active) simulation.alphaTarget(0); d.fx = event.x; d.fy = event.y; }));
svg.call(d3.zoom().scaleExtent([0.25, 4]).on('zoom', event => viewport.attr('transform', event.transform)));
if (data.unresolved.length) d3.select('body').append('div').attr('class', 'unresolved').text(`Unplaced configured entries: ${data.unresolved.join(', ')}`);
</script>
</body>
</html>
"""
NODE_COLORS = {
"commit": ("#d7f0ff", "#1f5f8b"),
"edge": ("#dff7df", "#2f6b2f"),
"replica": ("#fff1cc", "#8a6d1d"),
"proxy": ("#f4ddff", "#7b3f98"),
"broker": ("#ffe0cc", "#a64b00"),
}
SVG_P4_WIDTH = 330
SVG_HOST_WIDTH = 250
SVG_LINE_HEIGHT = 19
SVG_BOX_PADDING = 26
SVG_COLUMN_SPACING = 400
SVG_LAYER_SPACING = 360
SVG_HOST_OFFSET = 165
SVG_NODE_GAP = 70
SVG_MARGIN = 60
SVG_SCRIPT = r"""
var svgRoot = document.documentElement;
var nodes = {};
var edges = [];
var dragState = null;
var selection = [];
Array.prototype.forEach.call(svgRoot.querySelectorAll('g.node'), function (el) {
nodes[el.getAttribute('data-id')] = el;
});
Array.prototype.forEach.call(svgRoot.querySelectorAll('line.edge'), function (el) {
edges.push(el);
});
function addClass(el, name) {
var current = el.getAttribute('class') || '';
if (current.split(/\s+/).indexOf(name) === -1) {
el.setAttribute('class', (current + ' ' + name).trim());
}
}
function removeClass(el, name) {
var current = el.getAttribute('class') || '';
el.setAttribute('class', current.split(/\s+/).filter(function (part) {
return part && part !== name;
}).join(' '));
}
function clearSelection() {
selection.forEach(function (el) {
removeClass(el, 'selected');
});
selection = [];
}
function toggleSelection(el) {
var index = selection.indexOf(el);
if (index === -1) {
selection.push(el);
addClass(el, 'selected');
} else {
selection.splice(index, 1);
removeClass(el, 'selected');
}
}
function nodeCentre(el) {
var match = /translate\(\s*(-?[\d.]+)[ ,]+(-?[\d.]+)\s*\)/.exec(el.getAttribute('transform') || '');
return match ? { x: parseFloat(match[1]), y: parseFloat(match[2]) } : { x: 0, y: 0 };
}
function nodeHalfSize(el) {
return {
w: parseFloat(el.getAttribute('data-width')) / 2,
h: parseFloat(el.getAttribute('data-height')) / 2
};
}
function borderPoint(fromEl, toEl) {
var from = nodeCentre(fromEl);
var to = nodeCentre(toEl);
var half = nodeHalfSize(fromEl);
var dx = to.x - from.x;
var dy = to.y - from.y;
if (dx === 0 && dy === 0) {
return from;
}
var scale = Math.min(
Math.abs(dx) > 0.0001 ? half.w / Math.abs(dx) : Infinity,
Math.abs(dy) > 0.0001 ? half.h / Math.abs(dy) : Infinity
);
return { x: from.x + dx * scale, y: from.y + dy * scale };
}
function refreshEdges() {
edges.forEach(function (edge) {
var source = nodes[edge.getAttribute('data-source')];
var target = nodes[edge.getAttribute('data-target')];
if (!source || !target) {
return;
}
var start = borderPoint(source, target);
var end = borderPoint(target, source);
edge.setAttribute('x1', start.x.toFixed(1));
edge.setAttribute('y1', start.y.toFixed(1));
edge.setAttribute('x2', end.x.toFixed(1));
edge.setAttribute('y2', end.y.toFixed(1));
});
}
function toUserSpace(event) {
var point = svgRoot.createSVGPoint();
point.x = event.clientX;
point.y = event.clientY;
return point.matrixTransform(svgRoot.getScreenCTM().inverse());
}
svgRoot.addEventListener('mousedown', function (event) {
var group = event.target.closest ? event.target.closest('g.node') : null;
if (!group) {
clearSelection();
return;
}
if (event.shiftKey) {
toggleSelection(group);
event.preventDefault();
return;
}
if (selection.indexOf(group) === -1) {
clearSelection();
}
var moving = selection.length ? selection.slice() : [group];
dragState = {
pointer: toUserSpace(event),
items: moving.map(function (el) {
var centre = nodeCentre(el);
addClass(el, 'dragging');
return { el: el, x: centre.x, y: centre.y };
})
};
event.preventDefault();
});
svgRoot.addEventListener('mousemove', function (event) {
if (!dragState) {
return;
}
var pointer = toUserSpace(event);
var dx = pointer.x - dragState.pointer.x;
var dy = pointer.y - dragState.pointer.y;
dragState.items.forEach(function (item) {
item.el.setAttribute(
'transform',
'translate(' + (item.x + dx).toFixed(1) + ',' + (item.y + dy).toFixed(1) + ')'
);
});
refreshEdges();
});
svgRoot.addEventListener('mouseup', function () {
if (!dragState) {
return;
}
dragState.items.forEach(function (item) {
removeClass(item.el, 'dragging');
});
dragState = null;
});
function saveDiagram() {
clearSelection();
var fallback = svgRoot.getAttribute('data-filename') || 'p4_topology.svg';
var name = window.prompt('Save diagram as', fallback);
if (name === null) {
return;
}
name = name.trim() || fallback;
if (!/\.svg$/i.test(name)) {
name += '.svg';
}
var markup = new XMLSerializer().serializeToString(svgRoot);
var blob = new Blob([markup], { type: 'image/svg+xml;charset=utf-8' });
var url = URL.createObjectURL(blob);
var link = document.createElementNS('http://www.w3.org/1999/xhtml', 'a');
link.setAttribute('href', url);
link.setAttribute('download', name);
svgRoot.appendChild(link);
link.click();
svgRoot.removeChild(link);
URL.revokeObjectURL(url);
}
svgRoot.querySelector('#save-button').addEventListener('click', saveDiagram);
refreshEdges();
"""
def svg_border_point(source: dict, target: dict) -> tuple[float, float]:
dx = target["x"] - source["x"]
dy = target["y"] - source["y"]
if dx == 0 and dy == 0:
return source["x"], source["y"]
half_width = source["width"] / 2
half_height = source["height"] / 2
scale = min(
half_width / abs(dx) if abs(dx) > 1e-6 else float("inf"),
half_height / abs(dy) if abs(dy) > 1e-6 else float("inf"),
)
return source["x"] + dx * scale, source["y"] + dy * scale
def svg_layer_depths(servers: list[dict]) -> dict[str, int]:
by_id = {server["serverid"]: server for server in servers}
depths: dict[str, int] = {}
def depth_of(serverid: str, seen: frozenset[str]) -> int:
if serverid in depths:
return depths[serverid]
server = by_id.get(serverid)
target = server.get("target") if server else None
if not server or not target or target not in by_id or target in seen:
depths[serverid] = 0
else:
depths[serverid] = depth_of(target, seen | {serverid}) + 1
return depths[serverid]
for server in servers:
depth_of(server["serverid"], frozenset())
return depths
def build_svg_graph(servers: list[dict]) -> tuple[list[dict], list[dict]]:
nodes: list[dict] = []
nodes_by_id: dict[str, dict] = {}
for server in sorted(servers, key=lambda item: item["serverid"]):
lines = p4server_label_lines(server)
fill, stroke = NODE_COLORS[node_class(server["server_type"])]
node = {
"id": server["serverid"],
"kind": "p4",
"lines": lines,
"width": SVG_P4_WIDTH,
"height": SVG_BOX_PADDING + len(lines) * SVG_LINE_HEIGHT,
"fill": fill,
"stroke": stroke,
"x": 0.0,
"y": 0.0,
}
nodes.append(node)
nodes_by_id[node["id"]] = node
if server.get("host_target"):
continue
host_lines = host_label_lines(server)
host_node = {
"id": f"host-{server['serverid']}",
"kind": "host",
"lines": host_lines,
"width": SVG_HOST_WIDTH,
"height": SVG_BOX_PADDING + len(host_lines) * SVG_LINE_HEIGHT,
"fill": "#ffffff",
"stroke": "#66737c",
"x": 0.0,
"y": 0.0,
}
nodes.append(host_node)
nodes_by_id[host_node["id"]] = host_node
edges: list[dict] = []
for server in servers:
host_id = f"host-{server.get('host_target') or server['serverid']}"
if host_id in nodes_by_id:
edges.append({"source": server["serverid"], "target": host_id, "host": True})
if server.get("target") and server["target"] in nodes_by_id:
edges.append({"source": server["serverid"], "target": server["target"], "host": False})
return nodes, edges
def sugiyama_positions(nodes: list[dict], edges: list[dict]) -> dict[str, tuple[float, float]] | None:
try:
from fast_sugiyama import from_edges # pyright: ignore[reportMissingImports]
except ModuleNotFoundError:
return None
if not edges:
return None
try:
raw = from_edges([(edge["source"], edge["target"]) for edge in edges]).dot_layout().to_dict()
except Exception:
return None
# Dummy nodes for long edges come back with non-string keys.
known = {node["id"] for node in nodes}
placed = {key: value for key, value in raw.items() if key in known}
if len(placed) != len(known):
return None
widths = {node["id"]: node["width"] for node in nodes}
layers: dict[float, list[tuple[float, str]]] = {}
for key, (x, y) in placed.items():
layers.setdefault(y, []).append((x, key))
xs = [x for x, _ in placed.values()]
span = max(xs) - min(xs)
widest = max(len(row) for row in layers.values())
scale = (widest * SVG_COLUMN_SPACING) / span if span > 0 else 1.0
positions: dict[str, tuple[float, float]] = {}
# Roots come back with the highest y, so invert to read top-down.
for index, layer in enumerate(sorted(layers, reverse=True)):
y = index * SVG_LAYER_SPACING
previous_x: float | None = None
previous_width = 0.0
for x, key in sorted(layers[layer]):
scaled = x * scale
if previous_x is not None:
scaled = max(scaled, previous_x + (previous_width + widths[key]) / 2 + SVG_NODE_GAP)
positions[key] = (scaled, y)
previous_x, previous_width = scaled, widths[key]
return positions
def graphviz_positions(
nodes: list[dict], edges: list[dict], program: str = "dot"
) -> dict[str, tuple[float, float]] | None:
if program in GRAPHVIZ_FORCE_DIRECTED:
graph_attrs = [' overlap=prism;', ' sep="+40,40";']
elif program == "osage":
graph_attrs = [' nodesep=0.7;']
else:
graph_attrs = [" rankdir=TB;", " nodesep=0.7;", " ranksep=0.9;"]
source_lines = [
"digraph topology {",
*graph_attrs,
" node [shape=box, fixedsize=true];",
]
for node in nodes:
source_lines.append(
f' "{node["id"]}" [width={node["width"] / 72:.3f}, height={node["height"] / 72:.3f}];'
)
for edge in edges:
source_lines.append(f' "{edge["source"]}" -> "{edge["target"]}";')
source_lines.append("}")
try:
result = subprocess.run(
[program, "-Tplain"],
input="\n".join(source_lines),
capture_output=True,
text=True,
check=True,
)
except (FileNotFoundError, subprocess.CalledProcessError):
return None
positions: dict[str, tuple[float, float]] = {}
for line in result.stdout.splitlines():
if not line.startswith("node "):
continue
try:
parts = shlex.split(line)
except ValueError:
return None
positions[parts[1]] = (float(parts[2]) * 72, float(parts[3]) * 72)
known = {node["id"] for node in nodes}
if not known.issubset(positions):
return None
max_y = max(positions[key][1] for key in known)
return {key: (positions[key][0], max_y - positions[key][1]) for key in known}
def grid_positions(nodes: list[dict], servers: list[dict]) -> dict[str, tuple[float, float]]:
nodes_by_id = {node["id"]: node for node in nodes}
depths = svg_layer_depths(servers)
layers: dict[int, list[dict]] = {}
for server in servers:
layers.setdefault(depths[server["serverid"]], []).append(server)
positions: dict[str, tuple[float, float]] = {}
for depth in sorted(layers):
row = sorted(layers[depth], key=lambda item: item["serverid"])
for index, server in enumerate(row):
node = nodes_by_id[server["serverid"]]
x = SVG_P4_WIDTH / 2 + index * SVG_COLUMN_SPACING
y = node["height"] / 2 + depth * SVG_LAYER_SPACING
positions[node["id"]] = (x, y)
host_id = f"host-{server['serverid']}"
if host_id in nodes_by_id:
positions[host_id] = (x, y + SVG_HOST_OFFSET)
return positions
def layout_svg_nodes(nodes: list[dict], edges: list[dict], servers: list[dict], engine: str) -> str:
if engine == "auto":
candidates: tuple[str, ...] = ("sugiyama", "dot", "grid")
elif engine == "grid":
candidates = ("grid",)
else:
candidates = (engine, "grid")
positions = None
used = "grid"
for candidate in candidates:
if candidate == "sugiyama":
positions = sugiyama_positions(nodes, edges)
elif candidate in GRAPHVIZ_PROGRAMS:
positions = graphviz_positions(nodes, edges, candidate)
else:
positions = grid_positions(nodes, servers)
if positions is not None:
used = candidate
break
assert positions is not None
offset_x = SVG_MARGIN - min(positions[node["id"]][0] - node["width"] / 2 for node in nodes)
offset_y = SVG_MARGIN - min(positions[node["id"]][1] - node["height"] / 2 for node in nodes)
for node in nodes:
x, y = positions[node["id"]]
node["x"] = x + offset_x
node["y"] = y + offset_y
return used
def build_svg_nodes(servers: list[dict], engine: str = "auto") -> tuple[list[dict], list[dict]]:
nodes, edges = build_svg_graph(servers)
layout_svg_nodes(nodes, edges, servers, engine)
return nodes, edges
def render_svg(
title: str, data: dict, engine: str = "auto", filename: str = "p4_topology.svg"
) -> tuple[str, str]:
servers = list(iter_diagram_servers(data["servers"]))
nodes, edges = build_svg_graph(servers)
used_engine = layout_svg_nodes(nodes, edges, servers, engine)
nodes_by_id = {node["id"]: node for node in nodes}
width = max((node["x"] + node["width"] / 2 for node in nodes), default=600) + SVG_MARGIN
height = max((node["y"] + node["height"] / 2 for node in nodes), default=400) + SVG_MARGIN
parts = [
'<?xml version="1.0" encoding="UTF-8"?>',
f'<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" '
f'width="{width:.0f}" height="{height:.0f}" viewBox="0 0 {width:.0f} {height:.0f}" '
f'data-filename="{escape(filename)}">',
f" <title>{escape(title)}</title>",
" <defs>",
' <marker id="arrow" viewBox="0 -5 10 10" refX="9" refY="0" markerWidth="7" '
'markerHeight="7" orient="auto"><path d="M0,-5L10,0L0,5" fill="#5b6b78"/></marker>',
" </defs>",
" <style>",
" .node rect { stroke-width: 2px; rx: 8px; }",
" .node text { font-family: ui-sans-serif, system-ui, sans-serif; font-size: 13px; fill: #17212b; }",
" .node .heading { font-weight: 700; }",
" .node { cursor: move; }",
" .node.dragging rect { stroke-dasharray: 5 3; }",
" .node.selected rect { stroke-width: 4px; }",
" .edge { stroke: #5b6b78; stroke-width: 1.6px; marker-end: url(#arrow); }",
" .edge.host-edge { stroke: #9aa7b1; stroke-dasharray: 5 4; marker-end: none; }",
" #save-button { cursor: pointer; }",
" #save-button rect { fill: #17212b; }",
" #save-button text { fill: #ffffff; font-family: ui-sans-serif, system-ui, sans-serif; font-size: 13px; }",
" #hint { font-family: ui-sans-serif, system-ui, sans-serif; font-size: 12px; fill: #5b6b78; }",
" </style>",
f' <rect width="{width:.0f}" height="{height:.0f}" fill="#f4f7f9"/>',
" <g id=\"edges\">",
]
for edge in edges:
source = nodes_by_id[edge["source"]]
target = nodes_by_id[edge["target"]]
x1, y1 = svg_border_point(source, target)
x2, y2 = svg_border_point(target, source)
css_class = "edge host-edge" if edge["host"] else "edge"
parts.append(
f' <line class="{css_class}" data-source="{escape(edge["source"])}" '
f'data-target="{escape(edge["target"])}" x1="{x1:.1f}" y1="{y1:.1f}" '
f'x2="{x2:.1f}" y2="{y2:.1f}"/>'
)
parts.append(" </g>")
parts.append(' <g id="nodes">')
for node in nodes:
parts.append(
f' <g class="node {node["kind"]}" data-id="{escape(node["id"])}" '
f'data-width="{node["width"]}" data-height="{node["height"]:.0f}" '
f'transform="translate({node["x"]:.1f},{node["y"]:.1f})">'
)
parts.append(
f' <rect x="{-node["width"] / 2:.0f}" y="{-node["height"] / 2:.0f}" '
f'width="{node["width"]}" height="{node["height"]:.0f}" '
f'fill="{node["fill"]}" stroke="{node["stroke"]}"/>'
)
text_x = -node["width"] / 2 + 14
first_y = -node["height"] / 2 + SVG_BOX_PADDING
for index, line in enumerate(node["lines"]):
css_class = ' class="heading"' if index == 0 else ""
y = first_y + index * SVG_LINE_HEIGHT
parts.append(f' <text{css_class} x="{text_x:.0f}" y="{y:.0f}">{escape(line)}</text>')
parts.append(" </g>")
parts.append(" </g>")
parts.extend(
[
' <g id="save-button" transform="translate(20,20)">',
' <rect width="120" height="30" rx="6"/>',
' <text x="20" y="20">Save layout</text>',
" </g>",
' <text id="hint" x="152" y="40">Drag to move. Shift-click to select several, then drag together.</text>',
" <script type=\"application/ecmascript\"><![CDATA[" + SVG_SCRIPT + "]]></script>",
"</svg>",
]
)
return "\n".join(parts) + "\n", used_engine
def main() -> int:
args = parse_args()
if args.input_json:
data = load_data(args.input_json)
else:
data = collect_live_data(args.p4port, args.p4user)
save_json(args.json_output, data)
mermaid = render_mermaid(data)
write_text(args.mermaid_output, mermaid)
write_text(args.markdown_output, render_markdown(args.title, mermaid, data))
write_text(args.asciidoc_output, render_asciidoc(args.title, mermaid, data))
write_text(args.html_output, render_html(args.title, data))
svg_markup, svg_engine = render_svg(args.title, data, args.layout, args.svg_output.name)
write_text(args.svg_output, svg_markup)
if args.input_json and args.input_json != args.json_output:
save_json(args.json_output, data)
print(f"Wrote JSON data to {args.json_output}")
print(f"Wrote Mermaid diagram to {args.mermaid_output}")
print(f"Wrote Markdown wrapper to {args.markdown_output}")
print(f"Wrote AsciiDoc wrapper to {args.asciidoc_output}")
print(f"Wrote interactive HTML diagram to {args.html_output}")
print(f"Wrote editable SVG diagram to {args.svg_output} (layout: {svg_engine})")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except RuntimeError as exc:
print(str(exc), file=sys.stderr)
raise SystemExit(1)