Revit Pipe Creation with Fittings from 2D Linked CAD in Revit


Hi, I’m trying to create a script that uses python scripts to generate pipes in Revit from a 2D Linked CAD. However, only a certain number of pipes are being created, and a lot of the lines are being left out. I also noticed the issue may be with the collection of line segments as in CAD there are more segments (35 lines), but the dry run test only shows that it found 17-line segments for O2 Gas.

Here’s the code for the python script:

# =============================================================================
# Dynamo Python Script — Routed Pipes with Fittings from Linked CAD
# =============================================================================
#
# INPUTS — all set via Code Block nodes in Dynamo, no JSON file needed
#
#   IN[0]  CAD link element   — from Select Model Element / All Elements of Type
#   IN[1]  Layer names        — list of CAD layer name strings
#   IN[2]  System types       — list of Revit piping system type names
#   IN[3]  Pipe types         — list of Revit pipe type names
#   IN[4]  Level names        — list of Revit level names
#   IN[5]  Offsets mm         — list of heights above level in mm
#   IN[6]  Diameters mm       — list of fixed diameters, or 0 to read from CAD
#   IN[7]  Snap tolerance mm  — how close endpoints must be to merge (e.g. 25)
#   IN[8]  Clash clearance mm — min gap between pipes on diff layers (e.g. 50)
#   IN[9]  Dry run            — True = preview only, False = write to Revit
#
# OUTPUTS — use List.GetItemAtIndex nodes to read each one
#   index 0  summary / error   — counts, OR an error dict if something failed
#   index 1  system_type_log   — confirms system type assigned per layer
#   index 2  validation_errors — names that didn't match Revit
#   index 3  fitting_warnings  — junctions that failed or were skipped
#   index 4  clash_report      — cross-layer pipes below clearance threshold
#   index 5  pipe_elements     — the Revit pipe elements created
#
# IMPORTANT: This script ALWAYS sets OUT, even on failure, by wrapping
# everything in one master try/except. If something breaks, index 0 will
# contain {"error": "...", "traceback": "..."} instead of OUT being null.
# =============================================================================

import clr, math, traceback
clr.AddReference('RevitAPI')
clr.AddReference('RevitAPIUI')
clr.AddReference('RevitServices')

from Autodesk.Revit.DB import *
from Autodesk.Revit.DB.Plumbing import Pipe, PipeType, PipingSystemType
from RevitServices.Persistence import DocumentManager
from RevitServices.Transactions import TransactionManager
from collections import defaultdict

doc = DocumentManager.Instance.CurrentDBDocument

# Default failsafe output — overwritten only if everything succeeds
OUT = [
    {"error": "Script did not complete — see traceback below."},
    [], [], [], [], []
]

def unwrap_cad_link(raw):
    """Handles nested lists and Dynamo-wrapped elements."""
    val = raw
    while isinstance(val, list):
        if len(val) == 0:
            return None
        val = val[0]
    if hasattr(val, "InternalElement"):
        val = val.InternalElement
    elif hasattr(val, "UnwrapElement"):
        try:
            val = val.UnwrapElement()
        except:
            pass
    return val

def flatten(val):
    if isinstance(val, (list, tuple)):
        if len(val) == 1 and isinstance(val[0], (list, tuple)):
            return list(val[0])
        return list(val)
    return [val]

# =============================================================================
# MASTER TRY — guarantees OUT is always set, with full traceback on failure
# =============================================================================
try:

    # =========================================================================
    # 1 — READ INPUTS
    # =========================================================================
    cad_link       = unwrap_cad_link(IN[0])
    layer_names    = flatten(IN[1])
    system_names   = flatten(IN[2])
    pipetype_names = flatten(IN[3])
    level_names    = flatten(IN[4])
    offsets_mm     = flatten(IN[5])
    diameters_mm   = flatten(IN[6])
    snap_tol_mm    = IN[7] if IN[7] else 25
    clash_tol_mm   = IN[8] if IN[8] else 50
    dry_run        = bool(IN[9]) if IN[9] is not None else True

    if cad_link is None or not isinstance(cad_link, ImportInstance):
        OUT = [
            {"error": "IN[0] is not a valid CAD link (ImportInstance). "
                      "Click 'Select' on the Select Model Element node and "
                      "pick the linked CAD file in the Revit view. "
                      "Got type: %s" % type(cad_link)},
            [], [], [], [], []
        ]
        raise SystemExit()

    n_layers = len(layer_names)
    if n_layers == 0:
        OUT = [{"error": "IN[1] (layer names) is empty."}, [], [], [], [], []]
        raise SystemExit()

    if not all(len(x) == n_layers for x in [system_names, pipetype_names,
                                              level_names, offsets_mm, diameters_mm]):
        OUT = [
            {"error": "Input lists are different lengths. layers=%d systems=%d "
                      "pipetypes=%d levels=%d offsets=%d diameters=%d — "
                      "all must match." % (
                          n_layers, len(system_names), len(pipetype_names),
                          len(level_names), len(offsets_mm), len(diameters_mm))},
            [], [], [], [], []
        ]
        raise SystemExit()

    snap_tol  = float(snap_tol_mm)  / 304.8
    clash_tol = float(clash_tol_mm) / 304.8
    min_seg   = 10.0 / 304.8   # lowered to catch short connector segments
    tess_deg  = 5
    default_d = 100.0

    LW_TO_DIAM = {
        13:20, 18:25, 25:32, 35:40,  50:50,
        70:65, 100:80, 140:100, 200:125, 300:150, 400:200
    }
    COL_TO_DIAM = {
        1:25, 2:32, 3:40, 4:50, 5:65,
        6:80, 7:100, 8:125, 9:150, 30:200
    }

    # =========================================================================
    # 2 — REVIT LOOKUPS
    # =========================================================================
    def get_element_name(e):
        """
        Tries every known method to read a Revit element name.
        LookupParameter("Type Name") is used first — this is the most
        reliable method for PipeType and PipingSystemType across all
        Revit versions and project templates.
        """
        # Method 1 — LookupParameter "Type Name" (most reliable for system families)
        try:
            p = e.LookupParameter("Type Name")
            if p and p.AsString():
                return p.AsString()
        except: pass
        # Method 2 — Element.Name.GetValue static method
        try:
            n = Element.Name.GetValue(e)
            if n: return n
        except: pass
        # Method 3 — .Name property directly
        try:
            if e.Name: return e.Name
        except: pass
        return None

    def collect(cls):
        result = {}
        for e in FilteredElementCollector(doc).OfClass(cls).ToElements():
            nm = get_element_name(e)
            if nm:
                result[nm] = e
        return result

    all_pipe_types = collect(PipeType)
    all_sys_types  = collect(PipingSystemType)
    all_levels     = collect(Level)

    validation_errors = []
    system_type_log   = []

    def find(lookup, name, label):
        if name and name in lookup:
            return lookup[name]
        if name:
            for k, v in lookup.items():
                if k.lower() == name.lower():
                    return v
        validation_errors.append(
            "%s '%s' not found. Available: %s" % (label, name, sorted(lookup.keys()))
        )
        return list(lookup.values())[0] if lookup else None

    layer_configs = []
    for i in range(n_layers):
        name    = layer_names[i]
        sys_nm  = system_names[i]
        pt_nm   = pipetype_names[i]
        lv_nm   = level_names[i]
        off_mm  = float(offsets_mm[i])
        diam_mm = float(diameters_mm[i]) if diameters_mm[i] else 0.0

        pt = find(all_pipe_types, pt_nm,  "Pipe type   [%s]" % name)
        st = find(all_sys_types,  sys_nm, "System type [%s]" % name)
        lv = find(all_levels,     lv_nm,  "Level       [%s]" % name)

        system_type_log.append(
            "Layer: %-22s  System: %-28s  PipeType: %-15s  Level: %-15s  Offset: %gmm"
            % (name,
               get_element_name(st) if st else "NOT FOUND",
               get_element_name(pt) if pt else "NOT FOUND",
               get_element_name(lv) if lv else "NOT FOUND",
               off_mm)
        )

        layer_configs.append({
            "name":       name,
            "pipe_type":  pt,
            "sys_id":     st.Id if st else ElementId.InvalidElementId,
            "level":      lv,
            "offset_ft":  off_mm / 304.8,
            "fixed_diam": diam_mm if diam_mm > 0 else None,
        })

    layer_lookup = {lc["name"].upper(): lc for lc in layer_configs}

    # =========================================================================
    # 3 — DIAMETER RESOLUTION
    # =========================================================================
    def resolve_diam(prim, fixed):
        if fixed is not None:
            return float(fixed)
        try:
            lw = prim.LineWeight
            if lw in LW_TO_DIAM:
                return float(LW_TO_DIAM[lw])
        except: pass
        try:
            gs = doc.GetElement(prim.GraphicsStyleId)
            if gs:
                r = gs.GraphicsStyleCategory.LineColor.Red
                best_d, best_delta = default_d, 999
                for aci, d in COL_TO_DIAM.items():
                    delta = abs(r - aci * 28)
                    if delta < best_delta:
                        best_delta, best_d = delta, d
                if best_delta < 40:
                    return float(best_d)
        except: pass
        return default_d

    def layer_of(prim):
        try:
            gs = doc.GetElement(prim.GraphicsStyleId)
            return gs.GraphicsStyleCategory.Name if gs else ""
        except: return ""

    # =========================================================================
    # 4 — EXTRACT CURVES FROM CAD LINK
    # =========================================================================
    def arc_pts(arc, deg):
        try:
            sa, ea = arc.GetEndParameter(0), arc.GetEndParameter(1)
            total  = ea - sa
            steps  = max(2, int(math.ceil(abs(math.degrees(total)) / deg)))
            return [arc.Evaluate(sa + total * i / steps, False) for i in range(steps + 1)]
        except: return list(arc.Tessellate())

    opts = Options()
    opts.ComputeReferences = True
    raw_edges = defaultdict(list)

    cad_geometry = cad_link.get_Geometry(opts)
    if cad_geometry is None:
        OUT = [
            {"error": "CAD link returned no geometry. Check it's visible in "
                      "the current view and not hidden or unloaded."},
            [], [], [], [], [], []
        ]
        raise SystemExit()

    layers_seen_in_cad = set()

    def extract_primitives(geom_obj, xf_stack, depth=0):
        """
        Recursively unwrap GeometryInstance objects to find all
        Line/Arc primitives regardless of nesting depth.
        CAD files often have blocks inside blocks (2-3 levels deep).
        xf_stack accumulates the transforms at each level so we can
        convert local coordinates to Revit document coordinates.
        """
        if depth > 8:   # safety limit — no real CAD file nests deeper than this
            return

        if isinstance(geom_obj, GeometryInstance):
            # Accumulate this level's transform
            try:
                level_xf = geom_obj.Transform
                # Compose: apply parent transforms first, then this level
                combined = xf_stack.Multiply(level_xf) if xf_stack else level_xf
            except:
                combined = xf_stack

            # Try to get geometry with identity first (world coords), fall back
            try:
                children = geom_obj.GetInstanceGeometry(Transform.Identity)
                use_xf   = False   # coords already in world space
            except:
                children = geom_obj.GetInstanceGeometry()
                use_xf   = True    # coords in local space, need transform

            for child in children:
                if isinstance(child, GeometryInstance):
                    # Recurse into nested block
                    extract_primitives(child, combined if use_xf else Transform.Identity, depth + 1)
                else:
                    # It's a primitive — process it
                    process_primitive(child, combined if use_xf else Transform.Identity)

    def process_primitive(prim, xf):
        """Convert a single Line/Arc primitive to raw_edges entries."""
        lyr_up = layer_of(prim).upper()
        if lyr_up:
            layers_seen_in_cad.add(lyr_up)
        if lyr_up not in layer_lookup:
            return

        res   = layer_lookup[lyr_up]
        lev_z = (res["level"].Elevation if res["level"] else 0.0) + res["offset_ft"]
        diam  = resolve_diam(prim, res["fixed_diam"])

        def to_world(pt):
            try:
                p = xf.OfPoint(pt)
                return XYZ(p.X, p.Y, lev_z)
            except:
                return XYZ(pt.X, pt.Y, lev_z)

        def add_edge(p0, p1):
            s = to_world(p0)
            e = to_world(p1)
            if s.DistanceTo(e) >= min_seg:
                raw_edges[lyr_up].append((s, e, diam))

        if isinstance(prim, Line):
            add_edge(prim.GetEndPoint(0), prim.GetEndPoint(1))
        elif isinstance(prim, Arc):
            pts = arc_pts(prim, tess_deg)
            for i in range(len(pts) - 1):
                add_edge(pts[i], pts[i + 1])
        else:
            try:
                add_edge(prim.GetEndPoint(0), prim.GetEndPoint(1))
            except:
                pass

    # Iterate top-level geometry — extract_primitives handles all nesting
    for go in cad_geometry:
        extract_primitives(go, Transform.Identity)

    # Warn if none of the requested layers were actually found in the CAD file
    requested_upper = set(layer_lookup.keys())
    missing_layers = requested_upper - layers_seen_in_cad
    if missing_layers:
        for ml in missing_layers:
            validation_errors.append(
                "CAD layer '%s' was not found anywhere in the linked CAD geometry. "
                "Found layers include: %s"
                % (ml, sorted(list(layers_seen_in_cad))[:15])
            )

    # =========================================================================
    # 5 — BUILD NODE-EDGE NETWORKS
    # =========================================================================
    class Network:
        def __init__(self, tol):
            self.tol   = tol
            self.nodes = []
            self.edges = []

        def node(self, pt):
            for i, n in enumerate(self.nodes):
                if pt.DistanceTo(n) <= self.tol:
                    self.nodes[i] = XYZ((n.X+pt.X)/2, (n.Y+pt.Y)/2, (n.Z+pt.Z)/2)
                    return i
            self.nodes.append(pt)
            return len(self.nodes) - 1

        def add(self, p0, p1, d):
            i, j = self.node(p0), self.node(p1)
            if i != j: self.edges.append((i, j, d))

    networks = {}
    for lyr_up, segs in raw_edges.items():
        net = Network(snap_tol)
        for s, e, d in segs: net.add(s, e, d)
        networks[lyr_up] = net

    # =========================================================================
    # 6 — DRY-RUN REPORT
    # =========================================================================
    def make_dry_run():
        report = {
            "mode":                "DRY RUN — nothing written to Revit",
            "validation_errors":   validation_errors,
            "system_type_log":     system_type_log,
            "cad_layers_found":    sorted(list(layers_seen_in_cad))[:30],
            "layers":              []
        }
        for lyr_up, net in networks.items():
            res = layer_lookup[lyr_up]
            pc  = defaultdict(int)
            for ni, nj, _ in net.edges:
                pc[ni] += 1; pc[nj] += 1
            jc = defaultdict(int)
            for cnt in pc.values():
                if   cnt == 1: jc["dead_end"]       += 1
                elif cnt == 2: jc["elbow/straight"]  += 1
                elif cnt == 3: jc["tee"]             += 1
                elif cnt == 4: jc["cross"]           += 1
                else:          jc["complex_%d" % cnt]+= 1
            # Calculate ALL endpoint-to-endpoint distances to find real gap sizes
            gaps = []
            segs_for_layer = raw_edges.get(lyr_up, [])
            # Collect all unique endpoints from this layer
            all_endpoints = []
            for s, e, _ in segs_for_layer:
                all_endpoints.append(s)
                all_endpoints.append(e)
            # Measure every pair of endpoints
            for idx in range(len(all_endpoints)):
                for jdx in range(idx + 1, len(all_endpoints)):
                    d = all_endpoints[idx].DistanceTo(all_endpoints[jdx]) * 304.8
                    if 0.01 < d < 100000:
                        gaps.append(round(d, 1))
            gaps.sort()
            # Show smallest 10 gaps — the smallest non-zero ones
            # are the gaps between nearly-connected endpoints
            gap_sample = gaps[:10] if gaps else []

            report["layers"].append({
                "layer":            lyr_up,
                "pipe_type":        get_element_name(res["pipe_type"]) if res["pipe_type"] else "NOT FOUND",
                "level":            get_element_name(res["level"])     if res["level"]     else "NOT FOUND",
                "offset_mm":        round(res["offset_ft"] * 304.8),
                "segments":         len(net.edges),
                "nodes":            len(net.nodes),
                "junctions":        dict(jc),
                "snap_tol_used_mm": round(snap_tol * 304.8),
                "all_endpoint_gaps_mm_smallest_10": gap_sample,
                "tip": "Set snap to just above the SMALLEST gap value shown above to connect endpoints"
            })
        if not networks:
            report["warning"] = (
                "No pipe segments were extracted for ANY layer. Most likely cause: "
                "your layer names in IN[1] don't match the actual CAD layer names. "
                "Check 'cad_layers_found' above for the real names in this CAD file."
            )
        return report

    if dry_run:
        OUT = [make_dry_run(), system_type_log, validation_errors, [], [], []]
        raise SystemExit()

    # =========================================================================
    # 7 — CREATE PIPES
    # =========================================================================
    def set_p(elem, bip, val):
        p = elem.get_Parameter(bip)
        if p and not p.IsReadOnly:
            try: p.Set(val)
            except: pass

    def conn_near(pipe, pt):
        """
        Returns the closest open connector on a pipe to the given point.
        Uses a generous tolerance — the actual connector may be up to
        snap_tol away from the merged junction node centroid, so we
        search all connectors and return the nearest one unconditionally,
        then let the caller decide if it's close enough.
        """
        best, bd = None, 1e9
        for c in pipe.ConnectorManager.Connectors:
            d = c.Origin.DistanceTo(pt)
            if d < bd:
                bd, best = d, c
        return (best, bd) if best else (None, 1e9)

    def get_conns_for_junction(pipes, junction_pt):
        """
        For each pipe meeting at junction_pt, find the connector that is
        closest to the junction. Returns list of (connector, distance) pairs,
        one per pipe, sorted by distance ascending.
        Filters out connectors that are further than 2x snap_tol — anything
        beyond that isn't really at this junction.
        """
        result = []
        max_dist = snap_tol * 2.0   # generous — accounts for centroid shift
        for pipe in pipes:
            c, d = conn_near(pipe, junction_pt)
            if c and d < max_dist:
                result.append((c, d))
        return result

    has_fittings = len(list(
        FilteredElementCollector(doc)
        .OfCategory(BuiltInCategory.OST_PipeFitting)
        .WhereElementIsElementType().ToElements()
    )) > 0

    def insert_fitting(pt, pipes, fit_log, warn_log):
        """
        Inserts the appropriate fitting at a junction point.

        Key fixes vs previous version:
        1. conn_near now searches ALL connectors without a hard distance
           cutoff — the cutoff is applied here based on snap_tol instead
           of a fixed 0.08ft value that was too tight for merged junctions.
        2. For 2-pipe junctions, checks both collinear (no fitting needed)
           and angled (elbow needed) cases correctly.
        3. For tees, sorts connectors so the branch connector (the one
           not on the main run axis) is passed as the third argument to
           NewTeeFitting — Revit requires this specific order.
        4. Falls back to trying direct connector connection if fitting
           API fails — handles cases where pipes are already touching.
        """
        n = len(pipes)
        if n < 2: return

        conn_pairs = get_conns_for_junction(pipes, pt)
        conns = [cp[0] for cp in conn_pairs]

        if len(conns) < 2:
            warn_log.append(
                "SKIP: only %d/%d connectors found within %.0fmm of junction @ "
                "(%.0f,%.0f,%.0f)mm — pipes may not be physically touching"
                % (len(conns), n, snap_tol*304.8*2,
                   pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
            )
            return

        if not has_fittings:
            warn_log.append(
                "WARN: no pipe fitting families loaded in this project. "
                "Load fitting families in Revit then re-run. "
                "Junction @ (%.0f,%.0f,%.0f)mm skipped."
                % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
            )
            return

        try:
            if len(conns) == 2:
                # Check if pipes are collinear — if so, no fitting needed,
                # Revit joins them automatically
                d0 = conns[0].CoordinateSystem.BasisZ
                d1 = conns[1].CoordinateSystem.BasisZ
                dot = abs(d0.DotProduct(d1))
                if dot >= 0.9990:
                    # Collinear — attempt direct connector join instead
                    try:
                        conns[0].ConnectTo(conns[1])
                        fit_log.append(
                            "Joined (collinear) @ (%.0f,%.0f,%.0f)mm"
                            % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
                        )
                    except:
                        pass  # Already connected or zero-gap — fine
                    return
                # Angled — insert elbow
                doc.Create.NewElbowFitting(conns[0], conns[1])
                fit_log.append(
                    "Elbow @ (%.0f,%.0f,%.0f)mm"
                    % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
                )

            elif len(conns) >= 3:
                if len(conns) == 3:
                    # For tees, Revit expects: (run_end_1, run_end_2, branch)
                    # Identify the branch: it's the connector whose direction
                    # is most perpendicular to the other two
                    dirs = [c.CoordinateSystem.BasisZ for c in conns]
                    best_branch, best_score = 0, -1
                    for idx in range(3):
                        others = [dirs[j] for j in range(3) if j != idx]
                        # Branch has smallest dot product with the main run axis
                        score = 1.0 - abs(others[0].DotProduct(others[1]))
                        if score > best_score:
                            best_score, best_branch = score, idx
                    run_conns = [conns[i] for i in range(3) if i != best_branch]
                    branch    = conns[best_branch]
                    doc.Create.NewTeeFitting(run_conns[0], run_conns[1], branch)
                    fit_log.append(
                        "Tee @ (%.0f,%.0f,%.0f)mm"
                        % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
                    )

                elif len(conns) >= 4:
                    doc.Create.NewCrossFitting(conns[0], conns[1], conns[2], conns[3])
                    fit_log.append(
                        "Cross @ (%.0f,%.0f,%.0f)mm"
                        % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
                    )

        except Exception as ex:
            # If fitting API fails, try direct connector join as fallback
            try:
                if len(conns) >= 2 and not conns[0].IsConnected:
                    conns[0].ConnectTo(conns[1])
                    fit_log.append(
                        "Connected (fallback) @ (%.0f,%.0f,%.0f)mm"
                        % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8)
                    )
            except:
                pass
            warn_log.append(
                "FAIL @ (%.0f,%.0f,%.0f)mm: %s"
                % (pt.X*304.8, pt.Y*304.8, pt.Z*304.8, str(ex))
            )

    def pipe_env(pipe, extra):
        lc = pipe.Location
        if not isinstance(lc, LocationCurve): return None
        c = lc.Curve
        p0, p1, r = c.GetEndPoint(0), c.GetEndPoint(1), extra
        return (XYZ(min(p0.X,p1.X)-r, min(p0.Y,p1.Y)-r, min(p0.Z,p1.Z)-r),
                XYZ(max(p0.X,p1.X)+r, max(p0.Y,p1.Y)+r, max(p0.Z,p1.Z)+r), c, pipe)

    def overlap(a, b):
        return (a[0].X<=b[1].X and a[1].X>=b[0].X and
                a[0].Y<=b[1].Y and a[1].Y>=b[0].Y and
                a[0].Z<=b[1].Z and a[1].Z>=b[0].Z)

    all_created   = []
    all_fit_log   = []
    all_warn_log  = []
    all_skip_log  = []
    envs_by_layer = {}

    TransactionManager.Instance.EnsureInTransaction(doc)

    for lyr_up, net in networks.items():
        res     = layer_lookup[lyr_up]
        pt_id   = res["pipe_type"].Id if res["pipe_type"] else ElementId.InvalidElementId
        sys_id  = res["sys_id"]
        lev_id  = res["level"].Id if res["level"]     else ElementId.InvalidElementId
        off_ft  = res["offset_ft"]

        pipe_map    = {}
        layer_pipes = []

        for ni, nj, diam in net.edges:
            sp, ep = net.nodes[ni], net.nodes[nj]
            try:
                pipe = Pipe.Create(doc, sys_id, pt_id, lev_id, sp, ep)
                set_p(pipe, BuiltInParameter.RBS_PIPE_DIAMETER_PARAM, diam / 304.8)
                set_p(pipe, BuiltInParameter.RBS_OFFSET_PARAM, off_ft)
                pipe_map[(ni,nj)] = pipe_map[(nj,ni)] = pipe
                layer_pipes.append(pipe)
                all_created.append(pipe)
            except Exception as ex:
                all_skip_log.append("[%s] %d->%d: %s" % (lyr_up, ni, nj, str(ex)))

        node_pipes = defaultdict(list)
        for ni, nj, _ in net.edges:
            p = pipe_map.get((ni, nj))
            if p:
                if p not in node_pipes[ni]: node_pipes[ni].append(p)
                if p not in node_pipes[nj]: node_pipes[nj].append(p)

        fl, wl = [], []
        for ni, pipes in node_pipes.items():
            insert_fitting(net.nodes[ni], pipes, fl, wl)
        all_fit_log.extend(fl)
        all_warn_log.extend(wl)

        layer_envs = []
        for p in layer_pipes:
            try:
                r = (res["fixed_diam"] or default_d) / 304.8 / 2.0
                e = pipe_env(p, r + clash_tol)
                if e: layer_envs.append(e)
            except: pass
        envs_by_layer[lyr_up] = layer_envs

    TransactionManager.Instance.TransactionTaskDone()

    # =========================================================================
    # 8 — CLASH DETECTION
    # =========================================================================
    clash_report = []
    lkeys = list(envs_by_layer.keys())
    for i in range(len(lkeys)):
        for j in range(i + 1, len(lkeys)):
            la, lb = lkeys[i], lkeys[j]
            for ea in envs_by_layer[la]:
                for eb in envs_by_layer[lb]:
                    if not overlap(ea, eb): continue
                    try:
                        md = min(ea[2].Distance(eb[2].GetEndPoint(0)),
                                 ea[2].Distance(eb[2].GetEndPoint(1)),
                                 eb[2].Distance(ea[2].GetEndPoint(0)),
                                 eb[2].Distance(ea[2].GetEndPoint(1)))
                        if md < clash_tol:
                            clash_report.append({
                                "layer_a":      la,
                                "pipe_a_id":    ea[3].Id.IntegerValue,
                                "layer_b":      lb,
                                "pipe_b_id":    eb[3].Id.IntegerValue,
                                "clearance_mm": round(md * 304.8, 1)
                            })
                    except: pass

    # =========================================================================
    # 9 — OUTPUT (success path)
    # =========================================================================
    OUT = [
        {                                    # index 0 — summary
            "pipes_created":     len(all_created),
            "fittings_placed":   len(all_fit_log),
            "fitting_warnings":  len(all_warn_log),
            "clashes_detected":  len(clash_report),
            "edges_skipped":     len(all_skip_log),
            "validation_errors": len(validation_errors),
        },
        system_type_log,                     # index 1 — layer/system confirmation
        validation_errors,                   # index 2 — name mismatches
        all_warn_log,                        # index 3 — fitting warnings
        clash_report,                        # index 4 — cross-layer clashes
        all_created,                         # index 5 — pipe elements
        all_skip_log,                        # index 6 — skipped edges with reasons
    ]

except SystemExit:
    # Raised intentionally above (dry-run done, or a handled validation error)
    # OUT has already been set correctly — do nothing further.
    pass

except Exception as ex:
    # Catches EVERYTHING else, including Revit API / COM-level exceptions,
    # so OUT is never left null.
    OUT = [
        {
            "error": "Script crashed: %s" % str(ex),
            "traceback": traceback.format_exc()
        },
        [], [], [], [], []
    ]

It would be greatly appreciated if any advice or help was given, thanks!

Can you share a trimmed-down cad file? It appears to an issue on the cad file.

Hi, here’s a trimmed-down cad file with just the lines I wanted to create the pipes with some other lines.

TRIMMED CAD WITH ONLY LINES FOR PIPES.dwg (8.2 MB)

so i tested your file. but to be honest, it didn’t generate a single pipe, let alone 17. over 700 lines of code here, and from your coding style, it looks like you know what you’re doing. I feel that if you could narrow down the scope of the issue, it would be a lot easier to get more help. like which branch it executes or the exact exception it throws.

op’s input in DS, in case anyone wants to try:

layers = ["-ME-MG-PIPE-02", "-ME-MG-PIPE-CO2", "-ME-MG-PIPE-MA4"];
systems = ["O2 Gas", "CO2 Gas", "Medical Air 4"];
pipetypes = ["Copper Tube", "Copper Tube", "Copper Tube"];
levels = ["HOSPITAL ARAS 3","HOSPITAL ARAS 3","HOSPITAL ARAS 3"];
offsets = [3300,3300,3300];
diameters = [22,22,28];
snap = 1;
clash = 50;
dryrun = true;

Thanks for the input and trial, I managed to solve it. It was a problem with the extraction of layers from CAD file not the CAD file itself.

Gald you sorted it out

Could you share this Dynamo file? I’d like to learn from it.