Optimizing Revit API Script for Populating Conduit Network Parameters with Large Datasets

Revit
, , , ,
1

I am working on a Revit script to populate the “From Equipment Tag” and “To Equipment Tag” parameters for all elements in a conduit network. The process begins at a starting Junction Box and traverses through the connected conduits until reaching the end element, which could either be another Junction Box or a Light fixture.

The script currently iterates over 20 Junction Boxes to identify connected elements and populate the parameters accordingly. However, when the number of elements exceeds around 30, performance significantly degrades. With larger datasets (e.g., 300+ elements), Revit often becomes unresponsive due to the cumulative overhead of multiple API calls and filters.

Objective: Optimize the script to efficiently handle larger datasets without causing Revit to hang. Specifically, when collecting the clashing elements.

from RevitFunctions import *

import collections
import System
from System import Array
from System.Collections.Generic import List

import clr
clr.AddReference('ProtoGeometry')
from Autodesk.DesignScript.Geometry import *

clr.AddReference("RevitNodes")
import Revit
clr.ImportExtensions(Revit.Elements)
clr.ImportExtensions(Revit.GeometryConversion)

clr.AddReference("RevitServices")
import RevitServices
from RevitServices.Persistence import DocumentManager 
from RevitServices.Transactions import TransactionManager 

clr.AddReference("RevitAPI")
clr.AddReference("RevitAPIUI")

from Autodesk.Revit.DB import *
from Autodesk.Revit.UI import *



def collect_junction_boxes(document, keyword):
    if document:
        cat_filter = ElementCategoryFilter(BuiltInCategory.OST_ConduitFitting)
        if not keyword or keyword == "":
            TaskDialog.Show("Error", "Keyword not provided, collecting all the elements of specified category. Please note that it will impact performance.")
            return FilteredElementCollector(document).WherePasses(cat_filter).WhereElementIsNotElementType().ToElements()
        parameter_id = ElementId(BuiltInParameter.ELEM_FAMILY_AND_TYPE_PARAM)
        value_provider = ParameterValueProvider(parameter_id)
        evaluator = FilterStringContains()
        filter_rule = FilterStringRule(value_provider, evaluator, keyword)
        param_filter = ElementParameterFilter(filter_rule)
        combined_filter = LogicalAndFilter(cat_filter, param_filter)
        return FilteredElementCollector(document).WherePasses(combined_filter).WhereElementIsNotElementType().ToElements()
    TaskDialog.Show("Error", "Document can not be null.")
    return []
        

def get_clashing_conduit(document, main_element):
    if not document or not main_element:
        TaskDialog.Show("Error", "One or both of the args are null in get_clashing_conduit func")
        return None
    bb = main_element.get_BoundingBox(None)
    if not bb:
        # log it using logger class
        return None
    outline = Outline(bb.Min, bb.Max)
    bb_filter = BoundingBoxIntersectsFilter(outline)
    
    #category_list = List[BuiltInCategory]()
    #category_list.Add(BuiltInCategory.OST_Conduit)
    #category_list.Add(BuiltInCategory.OST_ConduitFitting)
    #cat_filter = ElementMulticategoryFilter(category_list)
    cat_filter = ElementCategoryFilter(BuiltInCategory.OST_Conduit)
    combined_filter = LogicalAndFilter(bb_filter, cat_filter)
    conduits = FilteredElementCollector(document).WherePasses(combined_filter).ToElements()
    if len(conduits) > 0:
        return conduits
    return None


def get_clashing_equipment(document, main_element):
    if not document or not main_element:
        TaskDialog.Show("Error", "One or both of the args are null in get_clashing_equipment func")
        return None
    bb = main_element.get_BoundingBox(None)
    if not bb:
        # log it using logger class
        return None
    outline = Outline(bb.Min, bb.Max)
    bb_filter = BoundingBoxIntersectsFilter(outline)
    cat_filter = ElementCategoryFilter(BuiltInCategory.OST_LightingFixtures)
    combined_filter = LogicalAndFilter(bb_filter, cat_filter)
    fixtures = FilteredElementCollector(document).WherePasses(combined_filter).ToElements()
    if len(fixtures) > 0:
        return fixtures[0]
    else:
        cat_filter = ElementCategoryFilter(BuiltInCategory.OST_ConduitFitting)
        combined_filter = LogicalAndFilter(bb_filter, cat_filter)
        fixtures = FilteredElementCollector(document).WherePasses(combined_filter).ToElements()
        if len(fixtures) > 0:
            for fixture in fixtures:
                if "jb" in fixture.Name.lower():
                    return fixture
    return None


def get_all_connected_elements(element, doc, include_owner=False, lookup=None):
    if not lookup:
        lookup = collections.OrderedDict()

    # Add the element itself if needed
    if include_owner and element.Id not in lookup:
        lookup[element.Id] = element

    # Determine connectors based on the element type
    connectors = None
    if hasattr(element, "ConnectorManager"):
        connectors = element.ConnectorManager.Connectors
    elif hasattr(element, "MEPModel"):
        connectors = element.MEPModel.ConnectorManager.Connectors
    else:
        return list(lookup.values())

    # Loop through connectors and find connected elements
    for connector in connectors:
        for ref in connector.AllRefs:
            # Skip self-references unless include_owner is True
            if ref.Owner.Id.Equals(element.Id) and not include_owner:
                continue
            elif isinstance(ref.Owner, MEPSystem):
                continue
            elif "JB" in doc.GetElement(ref.Owner.Id).Name:
                continue

            # Add the connected element if not already in lookup
            if ref.Owner.Id not in lookup:
                connected_elem = doc.GetElement(ref.Owner.Id)
                lookup[ref.Owner.Id] = connected_elem
                # Recurse to get elements connected to this new element
                get_all_connected_elements(connected_elem, doc, include_owner, lookup)

    return list(lookup.values())


doc = DocumentManager.Instance.CurrentDBDocument
uiapp = DocumentManager.Instance.CurrentUIApplication 
app = uiapp.Application 
uidoc = uiapp.ActiveUIDocument


all_junction_boxes = collect_junction_boxes(doc, "JB")
all_conduits = []
connected_elements = []
to_equipments = []

for i in range(20):
    junction_box = all_junction_boxes[i]
    conduits = get_clashing_conduit(doc, junction_box)
    all_conduits.append(conduits)
    
    temp_connected_elements = []
    temp_to_equipments = []
    
    for conduit in conduits:
        connected_elements_for_conduit = get_all_connected_elements(conduit, doc, include_owner=True)
        last_element = connected_elements_for_conduit[-1]
        to_equipment = get_clashing_equipment(doc, last_element)
        
        temp_connected_elements.append(connected_elements_for_conduit)
        temp_to_equipments.append(to_equipment)
    
    connected_elements.append(temp_connected_elements)
    to_equipments.append(temp_to_equipments)

transaction = Transaction(doc, "Conduit Tags From&TO")
transaction.Start()
# Loop to set equipment parameters
for junction_box, array_connected, list_of_to_equip in zip(all_junction_boxes, connected_elements, to_equipments):
    from_equipment_tag = junction_box.LookupParameter("Tag Number")
    
    for list_connected_elements, to_equipment in zip(array_connected, list_of_to_equip):
        to_equipment_tag = to_equipment.LookupParameter("Tag Number") if to_equipment else None
        
        for connected_element in list_connected_elements:
            from_equipment_param = connected_element.LookupParameter("From Equipment Tag")
            to_equipment_param = connected_element.LookupParameter("To Equipment Tag")
            
            if from_equipment_tag and from_equipment_tag.AsString() and from_equipment_param:
                from_equipment_param.Set(from_equipment_tag.AsString())
            if to_equipment_tag and to_equipment_tag.AsString() and to_equipment_param:
                to_equipment_param.Set(to_equipment_tag.AsString())
transaction.Commit()
# Output the results
OUT = all_junction_boxes, all_conduits, connected_elements, to_equipments

Edit: Removed some sensitive information.

image
image1149×740 41.1 KB

2

hi, I had to read for a while, I don’t really know where the categoryfilter was better based on the Id as it was faster
(should seek confirmation from people in the know)

categoryId=doc.Settings.Categories.get_Item(BuiltInCategory.OST_DuctAccessory).Id

CategoryFilter=ElementCategoryFilter(categoryId);

Sincerely
Christian.stan

3

Hi,
Junction boxes are not connected to the conduits ? , it would be preferable to use the connectors to retrieve each network.

3 Likes
4

At present it’s a No.

I am aware of this, and I’ve conveyed it to the team. The issue is that while they may connect these elements in Revit in the future, the site team still needs the data now for installation.

Anyway, I’ve created a workaround, and it’s functioning. I just need to add some final touches, after which I’ll share my approach for handling large datasets.

5

I am totally unaware about this🥲. But thanks for the reply.

1 Like
6

This works efficiently even with very large models, handling over 6,000 lights without performance issues.

import logging
import threading
import queue
import collections
import System
import getpass
from datetime import datetime
import os
from System import Array
from System.Collections.Generic import List

import clr

clr.AddReference('ProtoGeometry')
from Autodesk.DesignScript.Geometry import *

clr.AddReference("RevitNodes")
import Revit

clr.ImportExtensions(Revit.Elements)
clr.ImportExtensions(Revit.GeometryConversion)

clr.AddReference("RevitServices")
import RevitServices
from RevitServices.Persistence import DocumentManager
from RevitServices.Transactions import TransactionManager

clr.AddReference("RevitAPI")
clr.AddReference("RevitAPIUI")

from Autodesk.Revit.DB import *
from Autodesk.Revit.UI import *


def collect_junction_boxes(document, keyword):
    if not document:
        TaskDialog.Show("Error", "Document cannot be null.")
        return []

    # Filter category
    cat_filter = ElementCategoryFilter(BuiltInCategory.OST_ConduitFitting)
    collector = FilteredElementCollector(document).WherePasses(cat_filter).WhereElementIsNotElementType()

    # Apply keyword filter if provided
    if keyword:
        param_id = ElementId(BuiltInParameter.ELEM_FAMILY_AND_TYPE_PARAM)
        filter_rule = FilterStringRule(ParameterValueProvider(param_id), FilterStringContains(), keyword)
        collector = collector.WherePasses(ElementParameterFilter(filter_rule))
    else:
        TaskDialog.Show("Error",
                        "Keyword not provided. Collecting all elements in the category. This may impact performance.")

    return collector.ToElements()


def get_all_connected_elements(element, doc, include_owner=False, lookup=None):
    lookup = lookup or collections.OrderedDict()

    # Add the starting element if needed
    if include_owner:
        lookup.setdefault(element.Id, element)

    # Identify connectors
    connectors = (getattr(element, "ConnectorManager", None) or
                  getattr(getattr(element, "MEPModel", None), "ConnectorManager", None))
    if not connectors:
        return list(lookup.values())

    # Traverse through connectors to find connected elements
    for connector in connectors.Connectors:
        for ref in connector.AllRefs:
            ref_owner = ref.Owner
            if (ref_owner.Id.Equals(element.Id) and not include_owner) or isinstance(ref_owner, MEPSystem):
                continue
            if "JB" in doc.GetElement(ref_owner.Id).Name:
                continue

            # Add connected element and recurse
            if ref_owner.Id not in lookup:
                connected_elem = doc.GetElement(ref_owner.Id)
                lookup[ref_owner.Id] = connected_elem
                get_all_connected_elements(connected_elem, doc, include_owner, lookup)

    return list(lookup.values())


def are_bounding_boxes_clashing(bbox1, bbox2):
    # Check if there is overlap in the X direction
    x_overlap = (bbox1.Min.X <= bbox2.Max.X) and (bbox1.Max.X >= bbox2.Min.X)

    # Check if there is overlap in the Y direction
    y_overlap = (bbox1.Min.Y <= bbox2.Max.Y) and (bbox1.Max.Y >= bbox2.Min.Y)

    # Check if there is overlap in the Z direction
    z_overlap = (bbox1.Min.Z <= bbox2.Max.Z) and (bbox1.Max.Z >= bbox2.Min.Z)

    # If all three directions overlap, the bounding boxes are clashing
    return x_overlap and y_overlap and z_overlap


def junction_box_vs_conduit(document, cv_name):
    jb_vs_conduit_record = collections.OrderedDict()
    conduit_vs_jb_record = collections.OrderedDict()

    # Collect junction boxes and filter them by tag and CV name
    required_junction_boxes = [
        jb for jb in collect_junction_boxes(document, "JB")
        if jb.LookupParameter("Tag Number") and jb.LookupParameter("Tag Number").AsString()
           and (not cv_name or jb.LookupParameter("CV") and jb.LookupParameter("CV").AsString() == cv_name)
    ]

    # Get bounding boxes for junction boxes
    filtered_jbs = [(jb, jb.get_BoundingBox(None)) for jb in required_junction_boxes if jb.get_BoundingBox(None)]

    # Collect conduits with bounding boxes
    conduits = FilteredElementCollector(document).OfCategory(
        BuiltInCategory.OST_Conduit).WhereElementIsNotElementType().ToElements()
    filtered_conduits = [(conduit, conduit.get_BoundingBox(None)) for conduit in conduits if
                         conduit.get_BoundingBox(None)]

    # Check for clashing bounding boxes between junction boxes and conduits
    for jb, jb_bbox in filtered_jbs:
        for conduit, conduit_bbox in filtered_conduits:
            if are_bounding_boxes_clashing(jb_bbox, conduit_bbox):
                jb_vs_conduit_record.setdefault(jb.Id, []).append(conduit)
                conduit_vs_jb_record.setdefault(conduit.Id, jb)

    return jb_vs_conduit_record, conduit_vs_jb_record


def light_vs_conduit(document, cv_name):
    conduit_vs_light_record = collections.OrderedDict()

    # Filter lights by tag and CV
    all_lights = FilteredElementCollector(document).OfCategory(
        BuiltInCategory.OST_LightingFixtures).WhereElementIsNotElementType().ToElements()
    required_lights = [
        light for light in all_lights
        if light.LookupParameter("LIGHTING FIXTURE TAG") and light.LookupParameter("LIGHTING FIXTURE TAG").AsString()
           and (not cv_name or light.LookupParameter("CV") and light.LookupParameter("CV").AsString() == cv_name)
    ]

    # Get bounding boxes for required lights
    filtered_lights = [(light, light.get_BoundingBox(None)) for light in required_lights if light.get_BoundingBox(None)]

    # Collect conduits with bounding boxes
    conduits = FilteredElementCollector(document).OfCategory(
        BuiltInCategory.OST_Conduit).WhereElementIsNotElementType().ToElements()
    filtered_conduits = [(conduit, conduit.get_BoundingBox(None)) for conduit in conduits if
                         conduit.get_BoundingBox(None)]

    # Check for clashing bounding boxes between lights and conduits
    for light, light_bbox in filtered_lights:
        for conduit, conduit_bbox in filtered_conduits:
            if are_bounding_boxes_clashing(light_bbox, conduit_bbox):
                conduit_vs_light_record[conduit.Id] = light
                break  # Move to the next light once a clash is found for the current conduit

    return conduit_vs_light_record


def log_worker():
    while True:
        message = log_queue.get()
        if message == "STOP":
            break
        logger.info(message)
        log_queue.task_done()


def log_message(message):
    log_queue.put(message)


def create_temp_file(file_name):
    user_name = getpass.getuser()
    temp_dir = "C:\\Users\\{}\\AppData\\Local\\Temp".format(user_name)
    # time = datetime.now().strftime("%Hh-%Mm-%Ss")
    file_path = os.path.join(temp_dir, file_name + ".log")
    return file_path


doc = DocumentManager.Instance.CurrentDBDocument
cv_name = IN[0]

# Initialize logging
log_file_path = create_temp_file(doc.Title + "_Light_Conduit_Relationship")
logging.basicConfig(filename=log_file_path, level=logging.INFO, format='%(asctime)s - %(message)s')
logger = logging.getLogger()
log_queue = queue.Queue()

# Start logging thread
log_thread = threading.Thread(target=log_worker, daemon=True)
log_thread.start()

log_message("Execution Started")

# Generate clash records for junction boxes and conduits, lights and conduits
jb_vs_conduit, conduit_vs_jb = junction_box_vs_conduit(doc, cv_name)
conduit_vs_light = light_vs_conduit(doc, cv_name)
log_message("Clash data created")

# Start transaction
transaction = Transaction(doc, "Conduit Tags From&TO")
transaction.Start()

# Process each Junction Box and associated conduits
for jb_id, conduits in jb_vs_conduit.items():
    jb_element = doc.GetElement(jb_id)
    jb_tag = jb_element.LookupParameter("Tag Number").AsString() if jb_element else None
    log_message(f"Processing Junction Box ID: {jb_id.Value}, Tag: {jb_tag}")

    # Process each conduit connected to the current Junction Box
    for conduit in conduits:
        connected_elements = get_all_connected_elements(conduit, doc, include_owner=True)
        last_element = connected_elements[-1]

        # Determine "To Equipment Tag"
        equip_tag_param = None
        if last_element.Id in conduit_vs_light:
            equip_tag_param = conduit_vs_light[last_element.Id].LookupParameter("LIGHTING FIXTURE TAG")
        elif last_element.Id in conduit_vs_jb:
            equip_tag_param = conduit_vs_jb[last_element.Id].LookupParameter("Tag Number")

        equip_tag = equip_tag_param.AsString() if equip_tag_param else ""
        log_message(f"Conduit ID: {conduit.Id.Value}, To Equipment Tag: {equip_tag}")

        # Update connected elements' tags
        for elem in connected_elements:
            from_equipment_param = elem.LookupParameter("From Equipment Tag")
            to_equipment_param = elem.LookupParameter("To Equipment Tag")

            if jb_tag and from_equipment_param:
                from_equipment_param.Set(jb_tag)
                log_message(f"Set From Equipment Tag: {jb_tag} on element ID {elem.Id.Value}")

            if equip_tag and to_equipment_param:
                to_equipment_param.Set(equip_tag)
                log_message(f"Set To Equipment Tag: {equip_tag} on element ID {elem.Id.Value}")

# Commit transaction and stop logging thread
transaction.Commit()
log_message("Process completed")
log_queue.put("STOP")
log_thread.join()

# Output debug information
OUT = log_file_path
2 Likes