coulomb/railiance-fabric
Archived
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tune graph layout affinities

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-18 22:39:12 +02:00
parent b2165b5a2a
commit 5b6e4faf68
6 changed files with 349 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
railiance_fabric/graph_explorer.py
View File

@@ -1,12 +1,16 @@
from __future__ import annotations
from datetime import datetime, timezone
from re import sub
from typing import Any
from urllib.parse import urlparse
DISPLAY_STATES = ("show", "blur", "hide")
LAYER_ORDER = (
"repository",
"server",
"deployment",
"service",
"capability",
"interface",
@@ -17,6 +21,8 @@ LAYER_ORDER = (
_KIND_LAYER = {
"Repository": "repository",
"Server": "server",
"Deployment": "deployment",
"ServiceDeclaration": "service",
"CapabilityDeclaration": "capability",
"InterfaceDeclaration": "interface",
@@ -27,6 +33,8 @@ _KIND_LAYER = {
_LAYER_COLORS = {
"repository": "#475569",
"server": "#334155",
"deployment": "#16a34a",
"service": "#0f766e",
"capability": "#2563eb",
"interface": "#7c3aed",
@@ -42,6 +50,9 @@ _EDGE_STRENGTH = {
"consumes": "medium",
"uses_interface": "medium",
"declares": "weak",
"deployed_as": "medium",
"runs_on": "strong",
"owns_deployment": "weak",
}
@@ -85,7 +96,16 @@ def fabric_graph_explorer_manifest(base_url: str = "") -> dict[str, Any]:
}
for index, layer in enumerate(LAYER_ORDER)
],
"grouping_fields": ["domain", "repo", "layer", "kind", "lifecycle", "unresolved"],
"grouping_fields": [
"domain",
"repo",
"layer",
"kind",
"environment",
"serverHost",
"lifecycle",
"unresolved",
],
"search_fields": [
"id",
"stableKey",
@@ -94,6 +114,8 @@ def fabric_graph_explorer_manifest(base_url: str = "") -> dict[str, Any]:
"description",
"repo",
"domain",
"environment",
"serverHost",
"kind",
"layer",
"edgeType",
@@ -105,6 +127,8 @@ def fabric_graph_explorer_manifest(base_url: str = "") -> dict[str, Any]:
{"id": "layer", "label": "Layer", "type": "string"},
{"id": "repo", "label": "Repo", "type": "string"},
{"id": "domain", "label": "Domain", "type": "string"},
{"id": "environment", "label": "Environment", "type": "string"},
{"id": "serverHost", "label": "Server Host", "type": "string"},
{"id": "lifecycle", "label": "Lifecycle", "type": "string"},
{"id": "reviewState", "label": "Review State", "type": "string"},
{"id": "unresolved", "label": "Unresolved", "type": "boolean"},
@@ -196,10 +220,6 @@ def fabric_graph_explorer_payload(
repositories = repositories or []
snapshot_repo_slugs = snapshot_repo_slugs or set()
layers_by_id = {
str(node.get("id", "")): _layer_for_kind(str(node.get("kind", "")))
for node in source_nodes
}
source_repo_slugs = {
str(node.get("repo", "")).strip()
for node in source_nodes
@@ -298,53 +318,24 @@ def fabric_graph_explorer_payload(
}
)
edge_index = 0
node_layers = _node_data_index(elements, "layer")
node_repos = _node_data_index(elements, "repo")
edge_index = _append_infrastructure_elements(
source_nodes,
elements,
node_layers,
node_repos,
repo_slugs,
)
for edge in source_edges:
source = str(edge.get("from", ""))
target = str(edge.get("to", ""))
edge_type = str(edge.get("type", ""))
if not source or not target:
continue
source_layer = layers_by_id.get(source, "unknown")
target_layer = layers_by_id.get(target, "unknown")
strength = _edge_strength(edge_type)
edge_id = f"edge:{edge_index}:{source}:{edge_type}:{target}"
elements.append(_edge_element(edge_index, source, target, edge_type, node_layers, node_repos))
edge_index += 1
elements.append(
{
"data": {
"id": edge_id,
"stableKey": edge_id,
"kind": "edge",
"layer": "relationship",
"label": edge_type,
"source": source,
"target": target,
"sourceLayer": source_layer,
"targetLayer": target_layer,
"edgeType": edge_type,
"dependencyType": edge_type,
"strength": strength,
"edgeWidth": _edge_width(strength),
"sameLayer": source_layer == target_layer,
"reviewState": "accepted",
"freshnessState": "current",
"displayState": "show",
"visibilitySource": "default",
"visibilityReason": "default",
"deepLinks": {},
},
"classes": " ".join(
part
for part in (
edge_type.replace(":", "-"),
strength,
"same-layer" if source_layer == target_layer else "",
)
if part
),
}
)
for slug in sorted(repo_slugs):
repo_id = f"repo:{slug}"
@@ -352,36 +343,8 @@ def fabric_graph_explorer_payload(
node_id = str(node.get("id", ""))
if str(node.get("repo", "")) != slug or not node_id:
continue
edge_id = f"edge:{edge_index}:{repo_id}:declares:{node_id}"
elements.append(_edge_element(edge_index, repo_id, node_id, "declares", node_layers, node_repos))
edge_index += 1
target_layer = layers_by_id.get(node_id, "unknown")
elements.append(
{
"data": {
"id": edge_id,
"stableKey": edge_id,
"kind": "edge",
"layer": "relationship",
"label": "declares",
"source": repo_id,
"target": node_id,
"sourceLayer": "repository",
"targetLayer": target_layer,
"edgeType": "declares",
"dependencyType": "declares",
"strength": "weak",
"edgeWidth": 1,
"sameLayer": False,
"reviewState": "accepted",
"freshnessState": "current",
"displayState": "show",
"visibilitySource": "default",
"visibilityReason": "default",
"deepLinks": {},
},
"classes": "declares weak",
}
)
visible_nodes = [element for element in elements if "source" not in element["data"]]
visible_edges = [element for element in elements if "source" in element["data"]]
@@ -400,6 +363,8 @@ def fabric_graph_explorer_payload(
"blurred_count": 0,
"registered_repo_count": len(registered_repo_slugs),
"repo_node_count": len(repo_slugs),
"deployment_node_count": sum(1 for element in visible_nodes if element["data"].get("kind") == "Deployment"),
"server_node_count": sum(1 for element in visible_nodes if element["data"].get("kind") == "Server"),
"registered_only_repo_count": sum(
1
for element in visible_nodes
@@ -443,6 +408,268 @@ def _edge_width(strength: str) -> int:
return {"weak": 1, "medium": 3, "strong": 5}.get(strength, 2)
def _node_data_index(elements: list[dict[str, Any]], field: str) -> dict[str, str]:
index: dict[str, str] = {}
for element in elements:
data = element.get("data", {})
if isinstance(data, dict) and "source" not in data:
node_id = str(data.get("id") or "")
if node_id:
index[node_id] = str(data.get(field) or "")
return index
def _append_infrastructure_elements(
source_nodes: list[dict[str, Any]],
elements: list[dict[str, Any]],
node_layers: dict[str, str],
node_repos: dict[str, str],
repo_slugs: set[str],
) -> int:
edge_index = 0
endpoints_by_service = _endpoints_by_service(source_nodes)
server_ids_by_host: dict[str, str] = {}
service_nodes = sorted(
(node for node in source_nodes if node.get("kind") == "ServiceDeclaration"),
key=lambda node: str(node.get("id", "")),
)
for service in service_nodes:
service_id = str(service.get("id", ""))
if not service_id:
continue
attributes = service.get("attributes") if isinstance(service.get("attributes"), dict) else {}
environments = _environments(attributes)
repo = str(service.get("repo") or "")
service_name = str(service.get("name") or service_id)
service_endpoints = endpoints_by_service.get(service_id, [])
for environment in environments:
deployment_id = f"deployment:{service_id}:{environment}"
matching_endpoints = [
endpoint
for endpoint in service_endpoints
if _environment_matches(environment, endpoint["environments"])
]
server_hosts = sorted({endpoint["host"] for endpoint in matching_endpoints})
deployment_data = {
"id": deployment_id,
"stableKey": deployment_id,
"kind": "Deployment",
"layer": "deployment",
"label": f"{service_name} [{environment}]",
"name": f"{service_name} deployment ({environment})",
"description": f"{service_name} deployment in {environment}.",
"repo": repo,
"domain": str(service.get("domain") or ""),
"lifecycle": str(service.get("lifecycle") or ""),
"environment": environment,
"serviceId": service_id,
"serverHosts": server_hosts,
"reviewState": "accepted",
"freshnessState": "current",
"unresolved": False,
"confidence": 0.85 if server_hosts else 0.65,
"visualSize": 42,
"ownership": str(attributes.get("owner") or "repo"),
"attributes": {
"service_id": service_id,
"environment": environment,
"server_hosts": server_hosts,
"source_service": service_id,
},
"displayState": "show",
"visibilitySource": "default",
"visibilityReason": "default",
"sourceReferences": _source_references(service),
"deepLinks": {"service": f"/graph/nodes/{service_id}"},
}
elements.append({"data": deployment_data, "classes": "deployment accepted"})
node_layers[deployment_id] = "deployment"
node_repos[deployment_id] = repo
elements.append(_edge_element(edge_index, service_id, deployment_id, "deployed_as", node_layers, node_repos))
edge_index += 1
if repo and repo in repo_slugs:
repo_id = f"repo:{repo}"
elements.append(_edge_element(edge_index, repo_id, deployment_id, "owns_deployment", node_layers, node_repos))
edge_index += 1
for endpoint in matching_endpoints:
server_id = server_ids_by_host.get(endpoint["host"])
if server_id is None:
server_id = _server_id(endpoint["host"])
server_ids_by_host[endpoint["host"]] = server_id
server_data = {
"id": server_id,
"stableKey": server_id,
"kind": "Server",
"layer": "server",
"label": endpoint["host"],
"name": endpoint["host"],
"description": f"Server inferred from endpoint {endpoint['url']}.",
"repo": "",
"domain": str(service.get("domain") or ""),
"lifecycle": "active",
"environment": environment,
"serverHost": endpoint["host"],
"reviewState": "accepted",
"freshnessState": "current",
"unresolved": False,
"confidence": 0.7,
"visualSize": 48,
"ownership": "inferred",
"attributes": {
"host": endpoint["host"],
"source_interface_id": endpoint["interface_id"],
"endpoint_url": endpoint["url"],
},
"displayState": "show",
"visibilitySource": "default",
"visibilityReason": "default",
"sourceReferences": [{"label": "Endpoint interface", "ref": endpoint["interface_id"]}],
"deepLinks": {"interface": f"/graph/nodes/{endpoint['interface_id']}"},
}
elements.append({"data": server_data, "classes": "server accepted inferred"})
node_layers[server_id] = "server"
node_repos[server_id] = ""
elements.append(_edge_element(edge_index, deployment_id, server_id, "runs_on", node_layers, node_repos))
edge_index += 1
return edge_index
def _endpoints_by_service(source_nodes: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
endpoints: dict[str, list[dict[str, Any]]] = {}
for node in source_nodes:
if node.get("kind") != "InterfaceDeclaration":
continue
attributes = node.get("attributes") if isinstance(node.get("attributes"), dict) else {}
endpoint = attributes.get("endpoint") if isinstance(attributes.get("endpoint"), dict) else {}
url = str(endpoint.get("url") or "").strip()
host = _endpoint_host(url)
service_id = str(attributes.get("service_id") or "")
if not service_id or not host:
continue
endpoints.setdefault(service_id, []).append(
{
"host": host,
"url": url,
"interface_id": str(node.get("id") or ""),
"environments": _environments(attributes),
}
)
return endpoints
def _environments(attributes: dict[str, Any]) -> list[str]:
environments = [
str(environment)
for environment in attributes.get("environments", [])
if str(environment).strip()
]
return environments or ["all"]
def _environment_matches(deployment_environment: str, endpoint_environments: list[str]) -> bool:
return (
deployment_environment == "all"
or "all" in endpoint_environments
or deployment_environment in endpoint_environments
)
def _endpoint_host(url: str) -> str:
if not url:
return ""
parsed = urlparse(url)
host = parsed.netloc or parsed.path.split("/", 1)[0]
return host.strip().lower()
def _server_id(host: str) -> str:
key = sub(r"[^A-Za-z0-9._:+-]+", "-", host.lower()).strip("-")
return f"server:{key or 'unknown'}"
def _edge_element(
edge_index: int,
source: str,
target: str,
edge_type: str,
node_layers: dict[str, str],
node_repos: dict[str, str],
) -> dict[str, Any]:
source_layer = node_layers.get(source, "unknown")
target_layer = node_layers.get(target, "unknown")
source_repo = node_repos.get(source, "")
target_repo = node_repos.get(target, "")
same_repo = bool(source_repo and source_repo == target_repo)
strength = _edge_strength(edge_type)
layout = _layout_hints(edge_type, source_layer, target_layer, same_repo)
edge_id = f"edge:{edge_index}:{source}:{edge_type}:{target}"
return {
"data": {
"id": edge_id,
"stableKey": edge_id,
"kind": "edge",
"layer": "relationship",
"label": edge_type,
"source": source,
"target": target,
"sourceLayer": source_layer,
"targetLayer": target_layer,
"sourceRepo": source_repo,
"targetRepo": target_repo,
"edgeType": edge_type,
"dependencyType": edge_type,
"strength": strength,
"edgeWidth": _edge_width(strength),
"sameLayer": source_layer == target_layer,
"sameRepo": same_repo,
"layoutAffinity": layout["affinity"],
"layoutIdealLength": layout["ideal_length"],
"layoutElasticity": layout["elasticity"],
"reviewState": "accepted",
"freshnessState": "current",
"displayState": "show",
"visibilitySource": "default",
"visibilityReason": "default",
"deepLinks": {},
},
"classes": " ".join(
part
for part in (
edge_type.replace(":", "-"),
strength,
str(layout["affinity"]),
"same-layer" if source_layer == target_layer else "",
"same-repo" if same_repo else "",
)
if part
),
}
def _layout_hints(
edge_type: str,
source_layer: str,
target_layer: str,
same_repo: bool,
) -> dict[str, int | str]:
if edge_type == "runs_on":
return {"affinity": "deployment-server", "ideal_length": 42, "elasticity": 240}
if edge_type == "deployed_as":
return {"affinity": "service-deployment", "ideal_length": 58, "elasticity": 180}
if edge_type == "owns_deployment":
return {"affinity": "repo-deployment", "ideal_length": 150, "elasticity": 28}
if edge_type == "declares":
return {"affinity": "repo-declaration", "ideal_length": 88, "elasticity": 110}
if same_repo:
return {"affinity": "same-repo", "ideal_length": 72, "elasticity": 150}
if "repository" in {source_layer, target_layer}:
return {"affinity": "repo-loose", "ideal_length": 130, "elasticity": 45}
return {"affinity": "cross-repo", "ideal_length": 190, "elasticity": 22}
def _unresolved_dependency_ids(
nodes: list[dict[str, Any]],
edges: list[dict[str, Any]],