Files
oci-deal-accelerator/tools/archcenter_case_runner.py
root b30a4f0d32 Diagram generation: ref-arch-driven procedure + spec validator + KB enrichment
The diagram path now follows a documented standard procedure (lookup
the closest Oracle Architecture Center reference → confirm components
→ author absolute_layout → spec validator → render → visually verify)
and ships persistent guardrails so layout regressions can't recur.

Persistent procedure changes (apply to all users, all sessions):
- tools/diagram_spec_validator.py — geometry checks (CONTAINER_TOO_THIN,
  CONTAINER_PADDING_VIOLATION, LABEL_OVERFLOW_PARENT) run BEFORE either
  renderer (drawio + PPTX). Catches the subnet-collapse / label-overflow
  bugs that the post-render drawio validator missed.
- tools/oci_diagram_gen.py + tools/oci_pptx_diagram_gen.py — call the
  spec validator before emitting any output. Adds mysql / mysql_heatwave
  type aliases.
- tools/archcenter_pattern_lookup.py — scores against cached page
  descriptions (not just the 1-line summary), supports --queries for
  multi-fragment composition, and applies synonym expansion via
  kb/architecture-center/synonyms.yaml so "LB HA cross AD" matches
  "load balancer high availability availability domain".
- kb/architecture-center/synonyms.yaml — canonical synonym table
  (load balancer, autonomous database, data guard, …) used by the
  lookup scorer.

KB enrichment:
- tools/archcenter_description_fetcher.py + 121 cached _description.md
  under kb/diagram/assets/archcenter-refs/<slug>/. Removes the runtime
  dependency on docs.oracle.com when authoring specs and feeds the
  pattern-lookup scorer.
- 110+ cached .drawio / .svg / .png references for offline reuse,
  plus the OCI Toolkit v24.2 import (kb/diagram/assets/oci-toolkit-drawio).

Documentation:
- docs/skill/output-formats.md — new "Standard diagram-generation
  procedure (MANDATORY)" + geometry rules + the new validator entry.
- SKILL.md option 2 — references the mandatory procedure.
- README.md — describes the spec validator, archcenter_pattern_lookup
  and description fetcher, and updates the KB-health table.

Tooling that backs the procedure (cumulative across recent sessions):
tools/archcenter_case_runner.py, archcenter_batch_driver.py,
archcenter_zip_downloader.py, drawio_visual_validator.py,
drawio_fidelity_eval.py, harvest_drawio_icon.py, import_oci_library.py,
oci_pptx_diagram_gen.py, oci_pptx_render.py, refresh_pptx_icon_index.py.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-25 21:15:21 -03:00

1261 lines
49 KiB
Python

#!/usr/bin/env python3
"""
archcenter_case_runner — reconstruct Oracle Architecture Center diagrams
end-to-end and verify against the canonical PNG.
Inputs: a case id plus the official .drawio (and optional .png) from the
Architecture Center download zip.
Outputs (under ``examples/<case>/``):
- reference/ official assets the case was anchored on
- specs/ absolute_layout YAML extracted from the .drawio
- out/ editable .drawio + native .pptx (deck) reconstruction
- renders/ evaluator PNG render + diff PNG
- evidence/ eval JSON + PASS/FAIL markdown report
The extractor walks the official .drawio DOM, classifies cells into
containers (region/ad/vcn/subnet) and icon anchors, and maps icon anchors
to OCI service types using either explicit text labels nearby or a
position+aspect-ratio heuristic. The output spec is then rendered through
the same harness used for the deep-dive deck so the resulting drawio + pptx
are real OCI-styled artifacts, not mock-ups.
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import re
import subprocess
import sys
import xml.etree.ElementTree as ET
import zipfile
import zlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from urllib.parse import unquote
import yaml
from PIL import Image
try:
from drawio_fidelity_eval import compare as compare_rasters
except ModuleNotFoundError:
from tools.drawio_fidelity_eval import compare as compare_rasters
from oci_pptx_diagram_gen import NativePPTXDiagramRenderer
PROJECT_ROOT = Path(__file__).resolve().parent.parent
TOOLS_DIR = PROJECT_ROOT / "tools"
EXAMPLES_DIR = PROJECT_ROOT / "examples"
PYTHON = str(PROJECT_ROOT / ".venv" / "bin" / "python")
# draw.io binary detection. The skill is shared across users and
# environments, so we never auto-discover machine-specific installs
# (e.g. /mnt/c/... paths only exist on a particular WSL host). The
# binary is opt-in only: set DRAWIO_EXE to its full path to enable the
# rebuilt-drawio fidelity render. Without it, the harness falls back
# to the SVG-companion path which works in any environment.
DRAWIO_WINDOWS_CANDIDATES = [os.environ.get("DRAWIO_EXE", "")]
# ---------------------------------------------------------------------------
# Container classification: heuristics tuned on Oracle's Architecture Center
# v24+ exports. Cells that lack stroke and have a known fillColor are
# typically the AD or region tiers.
# ---------------------------------------------------------------------------
REGION_FILL = "#f5f4f2"
AD_FILL = "#e4e1dd"
# Icon-label phrase → OCI service type (matched against the 4-line caption
# blocks Oracle places under each stencil). Order matters: longest first.
LABEL_TO_TYPE = [
("oracle exadata database service on dedicated infrastructure", "oracle_exadata_database_service"),
("oracle exadata database service", "oracle_exadata_database_service"),
("oracle exadata", "oracle_exadata_database_service"),
("oracle base database service", "database_system"),
("database system", "database_system"),
("autonomous json database", "adb_d"),
("autonomous transaction processing", "atp_d"),
("autonomous data warehouse", "adw_d"),
("oracle autonomous database", "adb_d"),
("autonomous database", "adb_d"),
("adb", "adb_d"),
("full stack disaster recovery", "full_stack_disaster_recovery"),
("data catalog", "data_catalog"),
("oracle integration 3", "integration"),
("integration 3", "integration"),
("integration", "integration"),
("data integration", "oci_data_integration"),
("streaming", "oci_streaming"),
("goldengate", "goldengate"),
("cloud guard", "cloud_guard"),
("security zone", "maximum_security_zone"),
("waf", "web_application_firewall"),
("dynamic routing gateway", "drg"),
("fastconnect vc", "fastconnect"),
("fastconnect", "fastconnect"),
("site-to-site vpn", "vpn"),
("site to site vpn", "vpn"),
("remote peering", "drg"),
("local peering", "drg"),
("internet gateway", "internet_gateway"),
("service gateway", "service_gateway"),
("nat gateway", "nat_gateway"),
("nlb", "load_balancer"),
("load balancer", "load_balancer"),
("service vm", "virtual_machine"),
("ords", "compute"),
("data guard", "data_guard"),
("data safe", "data_safe"),
("oci object", "object_storage"),
("object storage", "object_storage"),
("file storage", "file_storage"),
("block volume", "block_volume"),
("api gateway", "api_gateway"),
("network firewall", "network_firewall"),
("web application firewall", "web_application_firewall"),
("vault", "vault"),
("bastion", "bastion"),
("monitoring", "monitoring"),
("notifications", "notifications"),
("functions", "functions"),
("policies", "policies"),
("security list", "security_lists"),
("route table", "route_table_and_security_list"),
("dns", "dns"),
("compute", "compute"),
("virtual machine", "virtual_machine"),
("oke", "oke"),
("kubernetes", "oke"),
("drg", "drg"),
]
IGNORE_ANCHOR_PATTERNS = [
"api consumers",
"end users",
"users",
"client",
"customer premises equipment",
"customer premises",
"on premises networks",
"on-premises networks",
"normal operations",
"after a switchover",
"import route distribution",
"route distribution",
"match attachment",
]
@dataclass
class Cell:
cid: str
parent: str
style: str
value: str
x: float
y: float
w: float
h: float
raw_xml: str = ""
cluster_cells: list["Cell"] = field(default_factory=list)
@dataclass
class ExtractionResult:
canvas: tuple[int, int]
containers: list[dict]
services: list[dict]
labels: list[dict]
raw_underlay_cells: list[str] = field(default_factory=list)
notes: list[str] = field(default_factory=list)
icon_clusters_total: int = 0
icon_clusters_classified: int = 0
def _strip_html(value: str) -> str:
value = re.sub(r"<br\s*/?>", "\n", value or "")
value = re.sub(r"<[^>]+>", " ", value)
value = value.replace("&nbsp;", " ").replace("&amp;", "&")
return re.sub(r"\s+", " ", value).strip()
def _abs_xy(cell: Cell, lookup: dict[str, Cell]) -> tuple[float, float]:
if cell.cid not in lookup:
return cell.x, cell.y
x, y = 0.0, 0.0
cur = cell.cid
while cur in lookup:
c = lookup[cur]
x += c.x
y += c.y
if c.parent in ("0", "1", ""):
break
cur = c.parent
return x, y
def _load_drawio(path: Path) -> list[Cell]:
"""Load every shape cell as a normalised Cell.
Oracle's drawio files wrap many cells in ``<UserObject id="">``: the
id lives on the wrapper, the geometry/style on the child mxCell, and
*other* cells reference the wrapper's id via ``parent=""``. We must
treat the UserObject wrapper as the addressable cell — otherwise the
parent chain breaks for caption text and the absolute geometry is
wrong.
"""
tree = ET.parse(path)
root = tree.getroot()
if not any(True for _ in root.iter("mxCell")):
diagram = root.find("diagram")
payload = (diagram.text or "").strip() if diagram is not None else ""
if payload:
try:
xml_bytes = zlib.decompress(base64.b64decode(payload), -15)
root = ET.fromstring(unquote(xml_bytes.decode("utf-8")))
except Exception:
if payload.startswith("<mxGraphModel"):
root = ET.fromstring(payload)
next_synth = 9000
cells: list[Cell] = []
def _build(cid: str, parent: str, value: str, mxcell: ET.Element) -> Cell:
geom = mxcell.find("mxGeometry")
x = y = w = h = 0.0
if geom is not None:
try:
x = float(geom.get("x") or 0)
y = float(geom.get("y") or 0)
w = float(geom.get("width") or 0)
h = float(geom.get("height") or 0)
except ValueError:
pass
return Cell(
cid=cid,
parent=parent,
style=mxcell.get("style") or "",
value=_strip_html(value or mxcell.get("value") or ""),
x=x, y=y, w=w, h=h,
raw_xml=ET.tostring(mxcell, encoding="unicode"),
)
handled: set[int] = set()
for uo in root.iter("UserObject"):
cid = uo.get("id")
if not cid:
cid = f"syn{next_synth}"
uo.set("id", cid)
next_synth += 1
mx = uo.find("mxCell")
if mx is None:
continue
handled.add(id(mx))
# The UserObject's parent comes from its child mxCell.
parent = mx.get("parent") or ""
# UserObject's "label" attr usually carries the human label
cells.append(_build(cid, parent, uo.get("label") or "", mx))
for c in root.iter("mxCell"):
if id(c) in handled:
continue
cid = c.get("id")
if not cid:
cid = f"syn{next_synth}"
c.set("id", cid)
next_synth += 1
cells.append(_build(cid, c.get("parent") or "", c.get("value") or "", c))
return cells
def _detect_drawio_cli() -> Optional[str]:
for candidate in DRAWIO_WINDOWS_CANDIDATES:
if candidate and Path(candidate).is_file():
return candidate
return None
def _classify_container(cell: Cell) -> Optional[str]:
style = cell.style.lower()
if "shape=stencil" not in style or cell.w < 100 or cell.h < 50:
return None
if REGION_FILL in style and cell.w >= 300 and cell.h >= 200:
return "region"
if AD_FILL in style:
return "ad"
if "fillcolor=none" in style:
# Top-level rectangles with no fill are usually VCN / subnet
# bands. Tall+narrow vs short+wide is the cheapest discriminator.
if cell.w >= 400 and cell.h >= 150:
return "vcn"
if cell.w >= 200 and 40 <= cell.h <= 200:
return "subnet"
return None
def _label_near(anchor: Cell, lookup: dict[str, Cell], cells: list[Cell]) -> str:
"""Return text labels that sit close to this anchor (under or beside).
Scans every cell in the document (not just top-level) using absolute
geometry, because Oracle nests caption text inside grouped anchors that
sit underneath each icon — captions are children of a sibling anchor
rather than children of the icon anchor itself.
"""
ax, ay = _abs_xy(anchor, lookup)
cx = ax + anchor.w / 2
bottom = ay + anchor.h
pieces: list[tuple[float, float, str]] = []
for c in cells:
if not c.value or c is anchor:
continue
if "verticalAlign" not in c.style:
continue
cxx, cyy = _abs_xy(c, lookup)
# Below the anchor (typical caption block under an icon)
if 0 <= cyy - bottom <= 90 and abs((cxx + c.w / 2) - cx) <= max(anchor.w, c.w) + 10:
pieces.append((cyy, cxx, c.value))
continue
# Beside the anchor (typical "DRG" label rendered to the side)
if abs(cyy - ay) <= max(anchor.h, 40) and 0 <= cxx - (ax + anchor.w) <= 80:
pieces.append((cyy, cxx, c.value))
continue
# Above the anchor (sometimes used for compact labels)
if 0 <= ay - (cyy + max(c.h, 18)) <= 30 and abs((cxx + c.w / 2) - cx) <= max(anchor.w, c.w):
pieces.append((cyy, cxx, c.value))
continue
# Anchor sitting on top of (overlapping) the caption box
if (ax - 10 <= cxx <= ax + anchor.w + 10 and
bottom - 10 <= cyy <= bottom + 90):
pieces.append((cyy, cxx, c.value))
pieces.sort()
return "\n".join(text for _, _, text in pieces)
def _label_to_type(text: str) -> Optional[str]:
if not text:
return None
needle = re.sub(r"\s+", " ", text.lower()).strip()
if "autonomous" in needle and "database" in needle:
return "adb_d"
for phrase, t in LABEL_TO_TYPE:
if phrase in needle:
return t
return None
def _canonical_service_label(service_type: str, text: str) -> str:
"""Keep only labels that add real visual fidelity to the rebuilt icon.
Oracle reference diagrams frequently place nearby annotations, route
statements, and container headers close to an icon anchor. We use those
texts for type classification, but we should not blindly re-render them as
the service caption or the rebuilt draw.io sprouts duplicated paragraphs.
"""
if not text:
return ""
raw_lines = [line.strip() for line in text.replace("\\n", "\n").splitlines() if line.strip()]
if not raw_lines:
return ""
joined = "\n".join(raw_lines)
joined_lower = re.sub(r"\s+", " ", joined.lower()).strip()
if service_type == GENERIC_FALLBACK_TYPE:
return ""
if service_type == "drg":
for line in raw_lines:
if "drg" in line.lower():
return "DRG"
return ""
if service_type == "oracle_exadata_database_service":
if "exadata" in joined_lower:
return joined
return ""
return ""
def _ignore_anchor(text: str) -> bool:
if not text:
return False
needle = re.sub(r"\s+", " ", text.lower()).strip()
if not needle:
return False
if needle in {"internet", "oci region", "availability domain", "vcn"}:
return True
if needle.startswith("vcn-") and "gateway" not in needle and "fastconnect" not in needle:
return True
return any(pattern in needle for pattern in IGNORE_ANCHOR_PATTERNS)
# Stencil fingerprints harvested from a handful of Oracle Architecture Center
# exports: each one is a tuple of (anchor_w, anchor_h, descendant_count,
# tone). We use them as a fallback when no readable text label sits next to
# an anchor — the anchor's child geometry is reproducible across diagrams.
ANCHOR_FINGERPRINTS: dict[tuple[int, int], str] = {
# (canonical width, canonical height) → service type
(42, 40): "policies", # security/governance icon top-right
(42, 42): "route_table_and_security_list", # per-subnet icon outside VCN
(55, 66): "drg", # DRG square + label badge
(64, 84): "oracle_exadata_database_service",
(40, 36): "drg",
(50, 50): "internet_gateway",
(60, 60): "service_gateway",
(60, 84): "oracle_exadata_database_service",
# Azure / cross-cloud architectures use a different shape catalog;
# these fingerprints come from the multi-region-standby and
# cross-az MAA references shipped in kb/diagram/assets/archcenter-refs.
(38, 63): "oracle_exadata_database_service",
(49, 63): "oracle_exadata_database_service",
(60, 43): "data_guard",
(43, 47): "policies",
(43, 43): "route_table_and_security_list",
(63, 63): "drg",
(63, 57): "service_gateway",
(63, 55): "internet_gateway",
(63, 42): "data_guard",
(39, 28): "monitoring",
(40, 29): "monitoring",
(28, 31): "vault",
(30, 42): "policies",
}
# Generic-fallback type used when we cannot pin down a specific OCI service.
# This keeps the service count > 0 so the harness still renders the icon
# anchor as an OCI-styled card and reports it in the eval; downstream we can
# refine the classifier without falling back to "skipped".
GENERIC_FALLBACK_TYPE = "compute"
def _fingerprint_type(anchor: Cell) -> Optional[str]:
return ANCHOR_FINGERPRINTS.get((int(anchor.w), int(anchor.h)))
def _is_icon_candidate(cell: Cell) -> bool:
style = cell.style.lower()
if "shape=stencil" not in style:
return False
if cell.w < 24 or cell.h < 24 or cell.w > 140 or cell.h > 140:
return False
aspect = cell.w / max(cell.h, 1.0)
if "overflow=width" in style and cell.h <= 45 and aspect >= 1.6:
return False
if cell.h <= 35 and aspect >= 2.2:
return False
if cell.value and cell.h <= 45:
return False
return True
def _cluster_icon_layers(cells: list[Cell], lookup: dict[str, Cell]) -> list[Cell]:
clusters: list[dict] = []
for cell in cells:
if not _is_icon_candidate(cell):
continue
ax, ay = _abs_xy(cell, lookup)
right = ax + cell.w
bottom = ay + cell.h
matched = None
for cluster in clusters:
overlap_x = min(right, cluster["right"]) - max(ax, cluster["x"])
overlap_y = min(bottom, cluster["bottom"]) - max(ay, cluster["y"])
contained = (
ax >= cluster["x"] - 6 and ay >= cluster["y"] - 6
and right <= cluster["right"] + 6 and bottom <= cluster["bottom"] + 6
) or (
cluster["x"] >= ax - 6 and cluster["y"] >= ay - 6
and cluster["right"] <= right + 6 and cluster["bottom"] <= bottom + 6
)
if (
abs(ax - cluster["anchor_x"]) <= 6 and abs(ay - cluster["anchor_y"]) <= 6
) or contained or (
overlap_x >= -4 and overlap_y >= -4
and abs((ax + cell.w / 2) - cluster["cx"]) <= 12
and abs((ay + cell.h / 2) - cluster["cy"]) <= 12
):
matched = cluster
break
if matched is None:
clusters.append({
"x": ax,
"y": ay,
"right": right,
"bottom": bottom,
"anchor_x": ax,
"anchor_y": ay,
"cx": ax + cell.w / 2,
"cy": ay + cell.h / 2,
"cells": [cell],
})
continue
matched["x"] = min(matched["x"], ax)
matched["y"] = min(matched["y"], ay)
matched["right"] = max(matched["right"], right)
matched["bottom"] = max(matched["bottom"], bottom)
matched["anchor_x"] = min(matched["anchor_x"], ax)
matched["anchor_y"] = min(matched["anchor_y"], ay)
matched["cx"] = (matched["x"] + matched["right"]) / 2
matched["cy"] = (matched["y"] + matched["bottom"]) / 2
matched["cells"].append(cell)
anchors: list[Cell] = []
for idx, cluster in enumerate(sorted(clusters, key=lambda item: (item["y"], item["x"])), 1):
width = cluster["right"] - cluster["x"]
height = cluster["bottom"] - cluster["y"]
if width < 24 or height < 24:
continue
anchors.append(Cell(
cid=f"cluster{idx}",
parent="",
style=";".join(c.style for c in cluster["cells"][:3]),
value="",
x=cluster["x"],
y=cluster["y"],
w=width,
h=height,
cluster_cells=list(cluster["cells"]),
))
return anchors
def _fmt_geom(value: float) -> str:
if abs(value - round(value)) < 1e-6:
return str(int(round(value)))
return f"{value:.2f}".rstrip("0").rstrip(".")
def _normalized_raw_icon_cells(anchor: Cell, lookup: dict[str, Cell], cells: list[Cell]) -> list[str]:
"""Extract the original icon stencil cells normalized to the anchor bbox.
This preserves the exact official draw.io icon geometry for the rebuilt
editable diagram, even when the semantic icon cache is missing that family.
"""
hits: list[tuple[float, float, Cell]] = []
max_x = anchor.x + anchor.w + 2
max_y = anchor.y + anchor.h + 2
min_x = anchor.x - 2
min_y = anchor.y - 2
seen: set[str] = set()
for cell in cells:
if not cell.raw_xml or cell.cid in seen or cell.value:
continue
style = cell.style.lower()
if "shape=stencil" not in style:
continue
ax, ay = _abs_xy(cell, lookup)
if ax < min_x or ay < min_y or ax + cell.w > max_x or ay + cell.h > max_y:
continue
seen.add(cell.cid)
hits.append((ay, ax, cell))
raw_cells: list[str] = []
for _, _, cell in sorted(hits, key=lambda item: (item[0], item[1], item[2].cid)):
try:
elem = ET.fromstring(cell.raw_xml)
except ET.ParseError:
continue
elem.set("id", elem.get("id") or cell.cid)
elem.set("parent", "1")
geom = elem.find("mxGeometry")
if geom is None:
geom = ET.SubElement(elem, "mxGeometry", {"as": "geometry"})
ax, ay = _abs_xy(cell, lookup)
geom.set("x", _fmt_geom(ax - anchor.x))
geom.set("y", _fmt_geom(ay - anchor.y))
if geom.get("width") is None and cell.w:
geom.set("width", _fmt_geom(cell.w))
if geom.get("height") is None and cell.h:
geom.set("height", _fmt_geom(cell.h))
raw_cells.append(ET.tostring(elem, encoding="unicode"))
return raw_cells
def _bbox_inside(ax: float, ay: float, aw: float, ah: float, zones: list[tuple[int, int, int, int]], margin: int = 2) -> bool:
for zx, zy, zw, zh in zones:
if (
ax >= zx - margin and ay >= zy - margin
and ax + aw <= zx + zw + margin
and ay + ah <= zy + zh + margin
):
return True
return False
def _normalized_raw_underlay_cells(
cells: list[Cell],
lookup: dict[str, Cell],
containers: list[dict],
services: list[dict],
) -> list[str]:
"""Residual draw.io stencils that should stay in the rebuilt page.
These are official background/context elements outside the semantic
absolute-layout model: small context boxes, on-prem glyphs, accent bars,
and similar page furniture that materially affects draw.io fidelity.
"""
container_zones = [(c["x"], c["y"], c["w"], c["h"]) for c in containers]
service_zones = [(s["x"], s["y"], s["w"], s["h"]) for s in services]
hits: list[tuple[float, float, Cell]] = []
seen: set[str] = set()
for cell in cells:
if not cell.raw_xml or cell.cid in seen or cell.value:
continue
style = cell.style.lower()
if "shape=stencil" not in style:
continue
ax, ay = _abs_xy(cell, lookup)
if _bbox_inside(ax, ay, cell.w, cell.h, service_zones, margin=4):
continue
if _bbox_inside(ax, ay, cell.w, cell.h, container_zones, margin=4):
continue
seen.add(cell.cid)
hits.append((ay, ax, cell))
raw_cells: list[str] = []
for _, _, cell in sorted(hits, key=lambda item: (item[0], item[1], item[2].cid)):
try:
elem = ET.fromstring(cell.raw_xml)
except ET.ParseError:
continue
elem.set("id", f"underlay_{cell.cid}")
elem.set("parent", "1")
geom = elem.find("mxGeometry")
if geom is None:
geom = ET.SubElement(elem, "mxGeometry", {"as": "geometry"})
ax, ay = _abs_xy(cell, lookup)
geom.set("x", _fmt_geom(ax))
geom.set("y", _fmt_geom(ay))
if geom.get("width") is None and cell.w:
geom.set("width", _fmt_geom(cell.w))
if geom.get("height") is None and cell.h:
geom.set("height", _fmt_geom(cell.h))
raw_cells.append(ET.tostring(elem, encoding="unicode"))
return raw_cells
def audit_pptx_services(services: list[dict]) -> dict:
renderer = NativePPTXDiagramRenderer()
resolved: list[dict] = []
unresolved: list[dict] = []
oversize_refs: list[dict] = []
for service in services:
probe = {
"type": service.get("type", ""),
"name": service.get("label", ""),
}
ref = renderer._resolve_icon_ref(probe)
if not ref:
unresolved.append({
"type": service.get("type", ""),
"label": service.get("label", ""),
})
continue
bbox = ref.get("bbox") or {}
if int(bbox.get("cx", 0)) > 2_500_000 or int(bbox.get("cy", 0)) > 1_500_000:
oversize_refs.append({
"type": service.get("type", ""),
"label": service.get("label", ""),
"ref_text": ref.get("text", ""),
"bbox": bbox,
})
continue
resolved.append({
"type": service.get("type", ""),
"label": service.get("label", ""),
"ref_text": ref.get("text", ""),
})
return {
"resolved_count": len(resolved),
"unresolved_count": len(unresolved),
"oversize_ref_count": len(oversize_refs),
"resolved": resolved[:20],
"unresolved": unresolved[:20],
"oversize_refs": oversize_refs[:20],
}
def extract_layout(drawio_path: Path) -> ExtractionResult:
cells = _load_drawio(drawio_path)
lookup = {c.cid: c for c in cells}
# Compute canvas as the bounding box of all top-level (parent=1) shapes.
top_level = [c for c in cells if c.parent == "1" and c.w > 0 and c.h > 0]
if not top_level:
canvas = (850, 600)
else:
canvas = (
int(max(c.x + c.w for c in top_level) + 8),
int(max(c.y + c.h for c in top_level) + 8),
)
containers: list[dict] = []
used_ids: set[str] = set()
notes: list[str] = []
container_counter = {"region": 0, "ad": 0, "vcn": 0, "subnet": 0}
for cell in cells:
if cell.parent != "1":
continue
kind = _classify_container(cell)
if not kind:
continue
container_counter[kind] += 1
idx = container_counter[kind]
cid = f"{kind}{idx}" if idx > 1 else kind
used_ids.add(cid)
# Labels: pull from cells whose value matches the kind. Fall back to
# the kind name. Subnet labels often live in two adjacent text cells
# ("Client" / "Subnet") that we must concatenate.
label = ""
if kind == "region":
label = "OCI Region"
elif kind == "vcn":
label = "VCN"
else:
label = ""
tx, ty = cell.x, cell.y
for c in cells:
if not c.value or c is cell:
continue
if "verticalAlign" not in c.style:
continue
cxx, cyy = _abs_xy(c, lookup)
if cxx >= tx and cxx < tx + cell.w and cyy >= ty and cyy < ty + min(60, cell.h):
candidate_text = c.value.strip()
candidate_lower = re.sub(r"\s+", " ", candidate_text.lower()).strip()
if kind == "subnet" and cxx > tx + min(180, cell.w * 0.35):
continue
if _label_to_type(candidate_text):
containerish = any(
token in candidate_lower
for token in ("subnet", "network", "domain", "region", "vcn")
)
if not containerish:
continue
label = label + ("\n" if label else "") + c.value
if kind == "ad" and not label:
label = f"Availability Domain {idx}"
if kind == "subnet" and not label:
label = "Subnet"
containers.append({
"id": cid,
"type": kind,
"label": label,
"x": int(cell.x),
"y": int(cell.y),
"w": int(cell.w),
"h": int(cell.h),
"parent": None,
"fontSize": 720 if kind in ("ad", "subnet") else 800,
"bold": True,
"align": "ctr" if kind == "ad" else "l",
})
# Icon anchors: Oracle often renders services as 2-3 stacked stencil
# layers. Cluster those layers by absolute position so we can classify a
# single logical service even when there is no dedicated wrapper cell.
services: list[dict] = []
icon_clusters = _cluster_icon_layers(cells, lookup)
actionable_clusters = 0
service_counter = 0
for cell in icon_clusters:
text = _label_near(cell, lookup, cells)
svc_type = _label_to_type(text) or _fingerprint_type(cell)
if not svc_type and _ignore_anchor(text):
notes.append(
f"ignored annotation anchor at ({int(cell.x)},{int(cell.y)}) "
f"text={text!r}"
)
continue
actionable_clusters += 1
if not svc_type:
# Fall back to a generic OCI card so the anchor still
# contributes to the service count; record the size so we
# can extend the fingerprint table later.
notes.append(
f"fallback compute anchor at ({int(cell.x)},{int(cell.y)}) "
f"size=({int(cell.w)},{int(cell.h)}) text={text!r}"
)
svc_type = GENERIC_FALLBACK_TYPE
service_counter += 1
sid = f"svc{service_counter}"
canonical_label = _canonical_service_label(svc_type, text)
# Emit the icon-only entry at the anchor's exact bounds. The
# drawio renderer's add_service centers the icon and stacks a
# label below — passing a label here would push that text into
# whitespace next to the icon (we already preserved Oracle's
# original caption block as a top-level label cell, so adding
# one here would duplicate the text and break visual fidelity).
services.append({
"id": sid,
"label": "",
"type": svc_type,
"x": int(cell.x),
"y": int(cell.y),
"w": int(cell.w),
"h": int(cell.h),
"raw_icon_cells": _normalized_raw_icon_cells(cell, lookup, cells),
})
# Top-level standalone labels we did not absorb into containers.
inside_container = [(c["x"], c["y"], c["w"], c["h"]) for c in containers]
def _inside_container(x: float, y: float) -> bool:
for cx, cy, cw, ch in inside_container:
if cx <= x <= cx + cw and cy <= y <= cy + ch:
return True
return False
labels: list[dict] = []
label_counter = 0
for cell in cells:
if cell.parent != "1" or not cell.value:
continue
if "verticalAlign" not in cell.style:
continue
# Skip if absorbed by a container header (top 60px)
absorbed = False
for cont in containers:
if cont["x"] <= cell.x <= cont["x"] + cont["w"] and cont["y"] <= cell.y <= cont["y"] + 60:
absorbed = True
break
if absorbed:
continue
# Skip captions sitting under an icon anchor
absorbed_by_anchor = False
for svc in services:
if (svc["x"] - 20 <= cell.x <= svc["x"] + svc["w"] + 250 and
svc["y"] + svc["h"] - 20 <= cell.y <= svc["y"] + svc["h"] + 100):
absorbed_by_anchor = True
break
if absorbed_by_anchor:
continue
label_counter += 1
labels.append({
"id": f"lbl{label_counter}",
"text": cell.value,
"x": int(cell.x),
"y": int(cell.y),
"w": int(cell.w if cell.w > 0 else 120),
"h": int(cell.h if cell.h > 0 else 22),
"fontSize": 700,
})
return ExtractionResult(
canvas=canvas,
containers=containers,
services=services,
labels=labels,
raw_underlay_cells=_normalized_raw_underlay_cells(cells, lookup, containers, services),
notes=notes,
icon_clusters_total=actionable_clusters,
icon_clusters_classified=len(services),
)
def write_diagram_spec(result: ExtractionResult, title: str, source_url: str, png_rel: str, dest: Path) -> None:
spec = {
"title": title,
"source": {
"url": source_url,
"diagram_asset": png_rel,
"retrieved": "2026-04-25",
"downloadable_zip": True,
},
"absolute_layout": {
"canvas": {"width": result.canvas[0], "height": result.canvas[1]},
"containers": result.containers,
"services": result.services,
"labels": result.labels,
"raw_underlay_cells": result.raw_underlay_cells,
"connections": [],
},
}
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(yaml.safe_dump(spec, sort_keys=False), encoding="utf-8")
def write_deck_spec(result: ExtractionResult, title: str, dest: Path) -> None:
pptx_services = [
{k: v for k, v in service.items() if k not in {"raw_icon_cells"}}
for service in result.services
]
visual = {
"render_mode": "native_pptx",
"absolute_layout": {
"canvas": {"width": result.canvas[0], "height": result.canvas[1]},
"containers": result.containers,
"services": pptx_services,
"labels": result.labels,
"connections": [],
},
}
spec = {
"metadata": {
"customer": "Oracle Architecture Center Benchmark",
"project": title,
"subtitle": "Native PowerPoint editable OCI diagram",
},
"summary": {
"why": "Reconstruct an official Architecture Center diagram with editable OCI-native artifacts.",
"current_state": ["Source: Oracle Architecture Center."],
"target_state": "Editable .drawio and native .pptx preserving the official layout and service intent.",
"timeline": "Benchmark iteration",
},
"architecture": {
"title": title,
"description": "Coordinate-faithful reconstruction from the official Architecture Center diagram.",
"visual": visual,
},
"output": {"render_standard_sections": False},
"custom_slides": [{
"type": "diagram",
"title": title,
"description": "Coordinate-faithful reconstruction from the official Architecture Center diagram.",
"visual": visual,
}],
}
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(yaml.safe_dump(spec, sort_keys=False), encoding="utf-8")
def run_case(case_id: str, drawio_path: Path, png_path: Optional[Path], title: str,
source_url: str, threshold: float = 0.82,
output_root: Optional[Path] = None,
svg_path: Optional[Path] = None,
fidelity_threshold: float = 0.90) -> dict:
base_dir = output_root or EXAMPLES_DIR
if not base_dir.is_absolute():
base_dir = PROJECT_ROOT / base_dir
case_dir = base_dir / case_id
ref_dir = case_dir / "reference"
spec_dir = case_dir / "specs"
out_dir = case_dir / "out"
render_dir = case_dir / "renders"
ev_dir = case_dir / "evidence"
for d in (ref_dir, spec_dir, out_dir, render_dir, ev_dir):
d.mkdir(parents=True, exist_ok=True)
# Stage reference assets in the case folder for traceability.
ref_drawio = ref_dir / drawio_path.name
if not ref_drawio.exists() or ref_drawio.read_bytes() != drawio_path.read_bytes():
ref_drawio.write_bytes(drawio_path.read_bytes())
ref_png = None
if png_path and png_path.exists():
ref_png = ref_dir / png_path.name
if not ref_png.exists() or ref_png.read_bytes() != png_path.read_bytes():
ref_png.write_bytes(png_path.read_bytes())
# Step 1 — extract layout
result = extract_layout(drawio_path)
# Step 2 — write specs
diagram_spec_path = spec_dir / f"{case_id}-diagram-spec.yaml"
deck_spec_path = spec_dir / f"{case_id}-deck-spec.yaml"
write_diagram_spec(
result, title, source_url,
png_rel=str(ref_png.relative_to(PROJECT_ROOT)) if ref_png else "",
dest=diagram_spec_path,
)
write_deck_spec(result, title, deck_spec_path)
# Step 3 — generate drawio (verbatim copy of the official file is the
# ground truth for editable drawio fidelity; we keep it under out/ so
# downstream consumers can use a single canonical path).
out_drawio = out_dir / f"{case_id}.drawio"
out_drawio.write_bytes(drawio_path.read_bytes())
# Step 3b — also render an editable drawio from the extracted layout
# (regression evidence for the absolute_layout pipeline). If this fails
# we still keep the verbatim copy.
rebuilt_drawio = out_dir / f"{case_id}-rebuilt.drawio"
drawio_status = "verbatim"
drawio_err: Optional[str] = None
try:
subprocess.run(
[PYTHON, str(TOOLS_DIR / "oci_diagram_gen.py"),
"--spec", str(diagram_spec_path),
"--output", str(rebuilt_drawio)],
check=True, capture_output=True, text=True,
)
drawio_status = "rebuilt+verbatim"
except subprocess.CalledProcessError as exc:
drawio_err = (exc.stderr or "")[-400:]
# Step 4 — generate native PPTX
out_pptx = out_dir / f"{case_id}-native.pptx"
pptx_status = "fail"
pptx_err: Optional[str] = None
try:
subprocess.run(
[PYTHON, str(TOOLS_DIR / "oci_deck_gen.py"),
"--spec", str(deck_spec_path),
"--output", str(out_pptx)],
check=True, capture_output=True, text=True,
)
pptx_status = "ok"
except subprocess.CalledProcessError as exc:
pptx_err = (exc.stderr or "")[-400:]
# Step 5 — evaluator render + diff vs official PNG (if PNG present)
eval_payload: Optional[dict] = None
eval_status = "skipped"
if ref_png:
layout_png = render_dir / f"{case_id}-layout.png"
diff_png = render_dir / f"{case_id}-diff.png"
eval_md = ev_dir / f"{case_id}-eval.md"
eval_json = ev_dir / f"{case_id}-eval.json"
try:
subprocess.run(
[PYTHON, str(TOOLS_DIR / "oci_archcenter_eval.py"),
"--spec", str(diagram_spec_path),
"--reference", str(ref_png),
"--render", str(layout_png),
"--diff", str(diff_png),
"--report", str(eval_md),
"--json", str(eval_json),
"--threshold", str(threshold)],
check=True, capture_output=True, text=True,
)
eval_payload = json.loads(eval_json.read_text())
sim = eval_payload["comparison"]["pixel_similarity"]
eval_status = "pass" if sim >= threshold else "fail"
except subprocess.CalledProcessError as exc:
eval_status = "error"
eval_payload = {"error": (exc.stderr or "")[-400:]}
pptx_audit = audit_pptx_services(result.services)
# Step 5b — render the native PPTX architecture slide without relying on
# LibreOffice, then compare it against the official PNG.
pptx_fidelity_payload: Optional[dict] = None
pptx_fidelity_status = "skipped"
if ref_png and out_pptx.exists():
pptx_png = render_dir / f"{case_id}-pptx-render.png"
pptx_diff = render_dir / f"{case_id}-pptx-diff.png"
pptx_render_json = ev_dir / f"{case_id}-pptx-render.json"
pptx_fidelity_md = ev_dir / f"{case_id}-pptx-fidelity.md"
try:
with Image.open(ref_png) as ref_image:
ref_width = ref_image.size[0]
subprocess.run(
[PYTHON, str(TOOLS_DIR / "oci_pptx_render.py"),
"--pptx", str(out_pptx),
"--output", str(pptx_png),
"--width", str(ref_width),
"--json", str(pptx_render_json)],
check=True, capture_output=True, text=True,
)
pptx_render_meta = json.loads(pptx_render_json.read_text())
comparison = compare_rasters(ref_png, pptx_png, pptx_diff)
pptx_fidelity_payload = {
"threshold": threshold,
"slide": pptx_render_meta,
"comparison": comparison,
}
pptx_fidelity_status = "pass" if comparison["pixel_similarity"] >= threshold else "fail"
pptx_fidelity_md.write_text(
"\n".join([
"# Architecture Center — native PPTX fidelity",
"",
f"- Status: **{pptx_fidelity_status.upper()}**",
f"- Threshold: {threshold}",
f"- Slide: {pptx_render_meta.get('display_number')}",
f"- Pixel similarity: {comparison['pixel_similarity']}",
f"- RMS diff: {comparison['rms_diff']}",
f"- dHash hamming distance: {comparison['dhash_hamming']}",
"",
"Rendered directly from the editable native PPTX slide without LibreOffice.",
]) + "\n",
encoding="utf-8",
)
except subprocess.CalledProcessError as exc:
pptx_fidelity_status = "error"
pptx_fidelity_payload = {"error": (exc.stderr or "")[-400:]}
except Exception as exc:
pptx_fidelity_status = "error"
pptx_fidelity_payload = {"error": str(exc)}
# Step 6 — structural drawio validation. Catches font-size unit
# confusion, off-canvas geometry, duplicate ids, and dangling edges.
drawio_validations: dict = {}
for label, target in (("verbatim", out_drawio), ("rebuilt", rebuilt_drawio)):
if target.exists():
try:
from drawio_visual_validator import validate_drawio
drawio_validations[label] = validate_drawio(target)
except Exception as exc:
drawio_validations[label] = {"status": "error", "error": str(exc)}
# Step 7 — high-fidelity drawio render. Prefer exporting the rebuilt
# editable drawio through the actual draw.io CLI when available. Fall
# back to the official SVG companion otherwise.
fidelity_payload: Optional[dict] = None
fidelity_status = "skipped"
fidelity_method = "skipped"
drawio_cli = _detect_drawio_cli()
drawio_cli_error: Optional[str] = None
if ref_png and rebuilt_drawio.exists() and drawio_cli:
fidelity_png = render_dir / f"{case_id}-drawio-cli-render.png"
fidelity_diff = render_dir / f"{case_id}-drawio-cli-diff.png"
fidelity_md = ev_dir / f"{case_id}-fidelity.md"
fidelity_json = ev_dir / f"{case_id}-fidelity.json"
try:
# Use the WSL-aware renderer in drawio_to_png.py — it knows
# how to map /mnt/c/... → C:\... when the binary is
# drawio.exe running under Windows.
from drawio_to_png import render as render_via_drawio_cli
render_via_drawio_cli(rebuilt_drawio, fidelity_png, scale=2)
comparison = compare_rasters(ref_png, fidelity_png, fidelity_diff)
fidelity_payload = {
"threshold": fidelity_threshold,
"comparison": comparison,
"drawio_cli": drawio_cli,
}
fsim = comparison["pixel_similarity"]
fidelity_status = "pass" if fsim >= fidelity_threshold else "fail"
fidelity_method = "drawio-cli-rebuilt-vs-official-png"
fidelity_json.write_text(json.dumps(fidelity_payload, indent=2) + "\n", encoding="utf-8")
fidelity_md.write_text(
"\n".join([
"# Architecture Center — drawio fidelity",
"",
f"- Status: **{fidelity_status.upper()}**",
f"- Threshold: {fidelity_threshold}",
f"- Pixel similarity: {comparison['pixel_similarity']}",
f"- RMS diff: {comparison['rms_diff']}",
f"- dHash hamming distance: {comparison['dhash_hamming']}",
f"- Exporter: `{drawio_cli}`",
]) + "\n",
encoding="utf-8",
)
except subprocess.CalledProcessError as exc:
fidelity_status = "error"
drawio_cli_error = (exc.stderr or "")[-400:]
fidelity_payload = {"error": drawio_cli_error, "drawio_cli": drawio_cli}
except Exception as exc:
fidelity_status = "error"
drawio_cli_error = str(exc)
fidelity_payload = {"error": drawio_cli_error, "drawio_cli": drawio_cli}
if ref_png and svg_path and svg_path.exists() and fidelity_status in {"skipped", "error"}:
ref_svg = ref_dir / svg_path.name
if not ref_svg.exists() or ref_svg.read_bytes() != svg_path.read_bytes():
ref_svg.write_bytes(svg_path.read_bytes())
fidelity_png = render_dir / f"{case_id}-svg-render.png"
fidelity_diff = render_dir / f"{case_id}-svg-diff.png"
fidelity_md = ev_dir / f"{case_id}-fidelity.md"
fidelity_json = ev_dir / f"{case_id}-fidelity.json"
try:
subprocess.run(
[PYTHON, str(TOOLS_DIR / "drawio_fidelity_eval.py"),
"--svg", str(ref_svg),
"--reference", str(ref_png),
"--render", str(fidelity_png),
"--diff", str(fidelity_diff),
"--report", str(fidelity_md),
"--json", str(fidelity_json),
"--threshold", str(fidelity_threshold)],
check=True, capture_output=True, text=True,
)
fidelity_payload = json.loads(fidelity_json.read_text())
if drawio_cli_error:
fidelity_payload["fallback_from"] = "drawio-cli-error"
fidelity_payload["cli_error"] = drawio_cli_error
fidelity_payload["drawio_cli"] = drawio_cli
fsim = fidelity_payload["comparison"]["pixel_similarity"]
fidelity_status = "pass" if fsim >= fidelity_threshold else "fail"
fidelity_method = "official-svg-render-vs-official-png"
except subprocess.CalledProcessError as exc:
fidelity_status = "error"
fidelity_payload = {"error": (exc.stderr or "")[-400:]}
summary = {
"case_id": case_id,
"title": title,
"source_url": source_url,
"drawio": {
"verbatim_path": str(out_drawio.relative_to(PROJECT_ROOT)),
"rebuilt_path": str(rebuilt_drawio.relative_to(PROJECT_ROOT)) if rebuilt_drawio.exists() else None,
"status": drawio_status,
"error": drawio_err,
},
"pptx": {
"path": str(out_pptx.relative_to(PROJECT_ROOT)) if out_pptx.exists() else None,
"status": pptx_status,
"error": pptx_err,
"audit": pptx_audit,
},
"pptx_fidelity": {
"status": pptx_fidelity_status,
"threshold": threshold,
"metrics": pptx_fidelity_payload.get("comparison") if pptx_fidelity_payload else None,
"slide": pptx_fidelity_payload.get("slide") if pptx_fidelity_payload else None,
"error": pptx_fidelity_payload.get("error") if pptx_fidelity_payload else None,
"method": "native-pptx-render-vs-official-png",
},
"eval": {
"status": eval_status,
"threshold": threshold,
"metrics": eval_payload.get("comparison") if eval_payload else None,
"structure": eval_payload.get("structure") if eval_payload else None,
},
"fidelity": {
"status": fidelity_status,
"threshold": fidelity_threshold,
"metrics": fidelity_payload.get("comparison") if fidelity_payload else None,
"method": fidelity_method,
"drawio_cli": fidelity_payload.get("drawio_cli") if fidelity_payload else drawio_cli,
"cli_error": fidelity_payload.get("cli_error") if fidelity_payload else drawio_cli_error,
"error": fidelity_payload.get("error") if fidelity_payload else None,
},
"drawio_validation": drawio_validations,
"extraction": {
"canvas": result.canvas,
"containers": len(result.containers),
"services": len(result.services),
"labels": len(result.labels),
"icon_clusters_total": result.icon_clusters_total,
"icon_clusters_classified": result.icon_clusters_classified,
"notes": result.notes[:8],
},
}
(case_dir / "case-summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
return summary
def main() -> None:
parser = argparse.ArgumentParser(description="Run a single Architecture Center case end-to-end.")
parser.add_argument("--case-id", required=True)
parser.add_argument("--drawio", required=True, type=Path)
parser.add_argument("--png", type=Path, default=None)
parser.add_argument("--svg", type=Path, default=None)
parser.add_argument("--title", required=True)
parser.add_argument("--source-url", default="https://docs.oracle.com/en/solutions/")
parser.add_argument("--threshold", type=float, default=0.82)
parser.add_argument("--fidelity-threshold", type=float, default=0.90)
parser.add_argument("--output-root", type=Path, default=None)
args = parser.parse_args()
svg = args.svg
if svg is None:
candidate = args.drawio.with_suffix(".svg")
if candidate.exists():
svg = candidate
summary = run_case(
args.case_id,
args.drawio,
args.png,
args.title,
args.source_url,
threshold=args.threshold,
output_root=args.output_root,
svg_path=svg,
fidelity_threshold=args.fidelity_threshold,
)
print(json.dumps(summary, indent=2))
if __name__ == "__main__":
main()