Script parametric blocks for BCAD

[schoeller]

Thx for PyRx.
Below script takes a directory of DWGs from a popular platform for pipe fittings, analyzes them (and the contained attribute definitions) and assembles a parametric blocks for use in BCAD. There seems to be an error somewhere hidden, since the flange glueing mechanism does not work yet. Still I wanted to share for others to possibly find inspiration. Coding style is influenced by Mistral AI.

Update 22.12.2025: component snapping is fixed

import traceback
import re
import math
import os
import logging
from pathlib import Path
from typing import List, Dict, Optional, Tuple, Union
import pandas as pd
from pyrx import Ap, Db, Ed, Ge, Rx, Brx

# --- Constants ---
DIRECTORY = r'C:\Users\SebastianSchoeller\Python\pyrx\Dev\createcomponent_1\dwg'
EXTENSION = '*.dwg'
PDMODE_VALUE = 32
PDSIZE_VALUE = 5
POINT_Z_OFFSET = 20
SIDE_1_Z_POSITION = 2
SIDE_2_Z_POSITION = -182

# Layers
LAYER_FLANGES = "STD_PARTS_FLANGES_CONNECTION"
LAYER_GASKETS = "STD_PARTS_GASKETS_CONNECTION"

# Constraints
CONSTRAINT_RIGID_SET_1 = "RigidSet_1"
CONSTRAINT_RIGID_SET_2 = "RigidSet_2"
CONSTRAINT_RIGID_SET_3 = "RigidSet_5"
CONSTRAINT_RIGID_SET_4 = "RigidSet_6"
CONSTRAINT_DISTANCE_1 = "Distance_3"
CONSTRAINT_DISTANCE_2 = "Distance_4"
CONSTRAINT_DISTANCE_3 = "Distance_7"
CONSTRAINT_DISTANCE_4 = "Distance_8"
CONSTRAINT_COORD_SYS = Brx.ConstraintArgCoordSysObj.eCSXY

# Defaults
DEFAULT_BLOCK_VARIABLES: Dict[str, Union[int, float]] = {
    'GasketThickness': 3,
    'A': 178,
    'Suppress': 1,
}

# --- Logging ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Helper Functions ---
def create_layer_if_missing(layer_table: Db.LayerTable, layer_name: str) -> None:
    """Create a layer if it doesn't exist."""
    if not layer_table.has(layer_name):
        layer_table.upgradeOpen()
        new_layer = Db.LayerTableRecord()
        new_layer.setName(layer_name)
        layer_table.add(new_layer)
        layer_table.downgradeOpen()
        logger.info(f"Created layer: {layer_name}")

def set_block_variables_if_missing(block_table_record_id: Db.ObjectId) -> None:
    """Set default values for block variables if they don't exist."""
    for name, value in DEFAULT_BLOCK_VARIABLES.items():
        var = Brx.Variable.getByName(block_table_record_id, name, True)
        var.setValue(value)

def create_point_and_line(
    model_space: Db.BlockTableRecord,
    point: Ge.Point3d,
    layer: str,
    direction: int,
) -> List[Db.Entity]:
    """Create a point and a line in the model space."""
    point_entity = Db.Point(point)
    point_entity.setLayer(layer)
    line_entity = Db.Line()
    line_entity.setStartPoint(point)
    line_entity.setEndPoint(point.transformBy(Ge.Matrix3d.translation(Ge.Vector3d(0, 0, POINT_Z_OFFSET) * direction)))
    line_entity.setLayer(layer)
    model_space.appendAcDbEntities([point_entity, line_entity])
    return [point_entity, line_entity]

def add_rigid_constraint_if_missing(
    constraint_group: Brx.ConstraintsGroup,
    entities: List[Db.Entity],
    constraint_name: str,
) -> None:
    """Add a rigid constraint to the specified entities if it doesn't exist."""
    if constraint_name not in {c.name() for c in constraint_group.constraints()}:
        paths = [Db.FullSubentPath(e.id(), Db.SubentId()) for e in entities]
        constraint = constraint_group.addConstraintSubents(Brx.ConstraintType.eAc3DRigidSet, paths)
        constraint.setName(constraint_name)

def add_distance_constraint_if_missing(
    constraint_group: Brx.ConstraintsGroup,
    entities: List[Db.Entity],
    constraint_name: str,
    coord_sys: Optional[Brx.ConstraintArgCoordSysObj] = None,
) -> None:
    """Add a distance constraint to the specified entities if it doesn't exist."""
    if constraint_name not in {c.name() for c in constraint_group.constraints()}:
        args = [Brx.ConstraintArgument(Db.FullSubentPath(e.id(), Db.SubentId())) for e in entities]
        if coord_sys:
            args.append(Brx.ConstraintArgument(coord_sys))
        constraint = constraint_group.addConstraintArgs(Brx.ConstraintType.eAc3DDistance, args)
        constraint.setName(constraint_name)

def erase_existing_ents_constraints(
    model_space: Db.BlockTableRecord,
    block_table_record_id: Db.ObjectId,
) -> None:
    """Erase existing entities and constraints."""
    for id in model_space.objectIds([Db.Point.desc(), Db.Line.desc(), Db.Solid3d.desc()]):
        ent = Db.Entity(id, Db.OpenMode.kForWrite)
        ent.erase()
    constraint_group = Brx.Core.getConstraintsGroup(block_table_record_id, True)
    for constraint in constraint_group.constraints():
        constraint_group.deleteConstraint(constraint)
    nid = Db.AssocNetwork.getInstanceFromObject(model_space.id(), False, False, "")
    if not nid.isNull():
        network = Db.AssocNetwork(nid)
        for _iterid in network.assocNetworkIterator():
            if _iterid.isDerivedFrom(Db.AssocVariable.desc()):
                try:
                    var = Db.AssocVariable(_iterid)
                    logger.info(f"Variable: {var.name()}, Expression: {var.expression()}, Value: {var.value()}")
                    if var.name().startswith("PartNumber"):
                        var.erase()
                except Exception as e:
                    logger.error(f"Error erasing variable: {e}")

def setup_geometry_and_constraints(database: Db.Database) -> None:
    """Set up geometry and constraints in the current database."""
    model_space = database.modelSpace(Db.OpenMode.kForWrite)
    block_table_record_id = database.currentSpaceId()
    constraint_group = Brx.Core.getConstraintsGroup(block_table_record_id, True)
    set_block_variables_if_missing(block_table_record_id)
    erase_existing_ents_constraints(model_space, block_table_record_id)
    constraint_group = Brx.Core.getConstraintsGroup(block_table_record_id, True)

    # Flanges
    side1_entities = create_point_and_line(model_space, Ge.Point3d(0, 0, SIDE_1_Z_POSITION), LAYER_FLANGES, -1)
    add_rigid_constraint_if_missing(constraint_group, side1_entities, CONSTRAINT_RIGID_SET_1)
    side2_entities = create_point_and_line(model_space, Ge.Point3d(0, 0, SIDE_2_Z_POSITION), LAYER_FLANGES, 1)
    add_rigid_constraint_if_missing(constraint_group, side2_entities, CONSTRAINT_RIGID_SET_2)
    add_distance_constraint_if_missing(constraint_group, [side1_entities[0]], CONSTRAINT_DISTANCE_1, CONSTRAINT_COORD_SYS)
    add_distance_constraint_if_missing(constraint_group, [side1_entities[0], side2_entities[0]], CONSTRAINT_DISTANCE_2)

    # Gaskets
    side1_entities = create_point_and_line(model_space, Ge.Point3d(0, 0, SIDE_1_Z_POSITION), LAYER_GASKETS, -1)
    add_rigid_constraint_if_missing(constraint_group, side1_entities, CONSTRAINT_RIGID_SET_3)
    side2_entities = create_point_and_line(model_space, Ge.Point3d(0, 0, SIDE_2_Z_POSITION), LAYER_GASKETS, 1)
    add_rigid_constraint_if_missing(constraint_group, side2_entities, CONSTRAINT_RIGID_SET_4)
    add_distance_constraint_if_missing(constraint_group, [side1_entities[0]], CONSTRAINT_DISTANCE_3, CONSTRAINT_COORD_SYS)
    add_distance_constraint_if_missing(constraint_group, [side1_entities[0], side2_entities[0]], CONSTRAINT_DISTANCE_4)

    constraint_group.evaluate()

def initialize_drawing() -> None:
    """Initialize the drawing with layers and default settings."""
    database = Db.curDb()
    Ed.Core.setVar("PDMODE", PDMODE_VALUE)
    Ed.Core.setVar("PDSIZE", PDSIZE_VALUE)
    layer_table = Db.LayerTable(database.layerTableId())
    create_layer_if_missing(layer_table, LAYER_FLANGES)
    create_layer_if_missing(layer_table, LAYER_GASKETS)
    setup_geometry_and_constraints(database)

def process_side_database(side_db: Db.Database, main_db: Db.Database) -> pd.DataFrame:
    """Process entities in the side database and return a DataFrame."""
    solid = Db.Solid3d()
    model_space = side_db.modelSpace()
    name, partnumber, size, rating, a = "", "", "", "", 0

    for attdef_id in model_space.objectIds(Db.AttributeDefinition.desc()):
        attdef = Db.AttributeDefinition(attdef_id)
        if attdef.tag() == 'PN':
            rating = f"PN {attdef.textString()}"
        elif attdef.tag() == 'DN':
            size = f"DN {attdef.textString()}"
        elif attdef.tag() == 'TXT01':
            name += attdef.textString()
        elif attdef.tag() == 'TXT02':
            partnumber = attdef.textString()
            partnumber = re.sub(r'(DN|PN)(?!\s)(\d+)', r'\1 \2', partnumber)
        elif attdef.tag() == 'TXT03':
            name += attdef.textString()
        elif attdef.tag() == 'L1':
            a = int(attdef.textString())

    for ref_id in model_space.objectIds(Db.BlockReference.desc()):
        block_ref = Db.BlockReference(ref_id)
        solids = [p for p in block_ref.explode() if p.isA() == Db.Solid3d.desc()]
        for ent in solids:
            ent.setColorIndex(256, True)
            solid.booleanOper(Db.BoolOperType.kBoolUnite, Db.Solid3d.cast(ent))

    solid.transformBy(Ge.Matrix3d.rotation(math.pi/2, Ge.Vector3d.kYAxis, Ge.Point3d(0, 0, 0)))
    solid.transformBy(Ge.Matrix3d.translation(Ge.Vector3d(0, 0, -float(a)/2)))

    df = pd.DataFrame({
        "Name": [name],
        "PartNumber": [partnumber],
        "Size": [size],
        "Rating": [rating],
        "A": [a],
    })
    df['objectid'] = main_db.addToModelspace(solid)
    return df

def open_and_process_dwg(dwg_path: str, processor) -> Optional[pd.DataFrame]:
    """Open a DWG file and process it with the given function."""
    side_db = Db.Database(False, True)
    try:
        side_db.readDwgFile(str(dwg_path))
        side_db.closeInput(True)
        return processor(side_db, Db.curDb())
    except Exception as e:
        logger.error(f"Error processing {dwg_path}: {e}")
        return None
    finally:
        side_db.closeInput(True)

def create_design_table(filename: str) -> None:
    """Create a reference design table for user parameters."""
    auto = Ed.AutoSysVar("CMDECHO", 0)
    resbuf = [
        (Rx.LispType.kText, "_DESIGNTABLE"),
        (Rx.LispType.kText, "F"),
        (Rx.LispType.kText, filename),
        (Rx.LispType.kNone, 0),
    ]
    Ed.Core.cmdS(resbuf)

def create_entity_suppression(id: Db.ObjectId, name: str) -> None:
    """Create an entity suppression."""
    ss = Ed.SelectionSet()
    ss.add(id)
    auto = Ed.AutoSysVar("CMDECHO", 0)
    resbuf = [
        (Rx.LispType.kText, "_SUPPRESS"),
        (Rx.LispType.kText, "_N"),
        (Rx.LispType.kText, name),
        (Rx.LispType.kSelectionSet, ss),
        (Rx.LispType.kText, ""),
        (Rx.LispType.kNone, 0),
    ]
    Ed.Core.cmdS(resbuf)

def set_var_value(var: Db.AssocVariable, expression: str) -> Db.AssocVariable:
    """Set the value of an associative variable."""
    var.upgradeOpen()
    var.setExpression(expression, "AcDbCalc:1.0", True, True, False)
    var.downgradeOpen()
    return var

def modify_assocs(db: Db.Database, name: str, index: int, length: int) -> None:
    """Modify associative variables."""
    nid = Db.AssocNetwork.getInstanceFromObject(db.modelSpaceId(), False, False, "")
    if nid.isNull():
        raise RuntimeError("Associative network not found.")
    network = Db.AssocNetwork(nid)
    for _iterid in network.assocNetworkIterator():
        if not _iterid.isDerivedFrom(Db.AssocVariable.desc()):
            continue
        var = Db.AssocVariable(_iterid)
        if var.name() == name:
            set_var_value(var, f'abs({index}-Suppress)')
        elif var.name() in {CONSTRAINT_DISTANCE_1, CONSTRAINT_DISTANCE_3}:
            set_var_value(var, 'GasketThickness/2')
        elif var.name() in {CONSTRAINT_DISTANCE_2, CONSTRAINT_DISTANCE_4}:
            set_var_value(var, 'A+GasketThickness')

def expose_variables(db: Db.Database) -> None:
    """Expose selected variables."""
    for var_name, expose_mode in [
        ('PartNumber', Brx.VariableExposeMode.eOn),
        ('A', Brx.VariableExposeMode.eOff),
        ('Suppress', Brx.VariableExposeMode.eOff),
        ('Size', Brx.VariableExposeMode.eOn),
        ('Rating', Brx.VariableExposeMode.eOn),
    ]:
        var = Brx.Variable.getByName(db.currentSpaceId(), var_name, False)
        if var:
            var.setExposed(expose_mode)

def simplify_geometry() -> None:
    """Simplify all geometry."""
    auto = Ed.AutoSysVar("CMDECHO", 0)
    resbuf = [
        (Rx.LispType.kText, "_DMSIMPLIFYALL"),
        (Rx.LispType.kText, "E"),
        (Rx.LispType.kNone, 0),
    ]
    Ed.Core.cmdS(resbuf)

def finalize_drawing(data: pd.DataFrame) -> None:
    """Finalize the drawing and save the output."""
    database = Db.curDb()
    data['Suppress'] = data.index + 1
    filename = f"{data.iloc[0, 0]}.csv"
    selected_columns = ["PartNumber", "Size", "Rating", "A", "Suppress"]
    data[selected_columns].to_csv(filename, header=True, index=False)
    logger.info(f"Saved to {filename}")
    create_design_table(filename)
    for _, row in data.iterrows():
        create_entity_suppression(Db.Solid3d(row['objectid']).id(), str(row['PartNumber']).replace(" ", "_"))
        modify_assocs(database, str(row['PartNumber']).replace(" ", "_"), row['Suppress'], len(data))
    simplify_geometry()
    expose_variables(database)
    output_filename = f"{data.iloc[0, 0]}.dwg"
    database.saveAs(os.path.join(os.getcwd(), output_filename), False, Db.DwgVersion.kDHL_CURRENT)
    logger.info(f"Drawing saved as {output_filename}")

# --- Main Command ---
@Ap.Command()
def doit() -> None:
    """Main command to initialize and process DWG files."""
    try:
        rows = []
        initialize_drawing()
        for dwg_file in Path(DIRECTORY).glob(EXTENSION):
            logger.info(f"Processing {dwg_file}")
            row = open_and_process_dwg(str(dwg_file), process_side_database)
            if row is not None:
                rows.append(row)
        if rows:
            df = pd.concat(rows, ignore_index=True)
            finalize_drawing(df)
        logger.info("Done!")
    except Exception as err:
        logger.error(f"Error in doit: {err}")
        traceback.print_exception(err)

VAG EKN Butterfly ValveEpoxy - R14 - Electric Actuator (EA).zip