Python Node Exception Handling

See my python script below for setting pressure pipes to a reference surface in Civil 3D. I think I am close with this script, I just can’t figure out how to handle the PointNotOnEntity exception without the script stopping. I have tried try, except blocks, but that doesn’t seem to do the trick. I ran into an issue with something similar for alignments, but the alignments had a specific method that allows for accepting out of bounds values.

Any help in handling these exception and keeping the script running would be much appreciated.

# Load the Python Standard and DesignScript Libraries
import sys
import clr

# Add Assemblies for AutoCAD and Civil3D
clr.AddReference('AcMgd')
clr.AddReference('AcCoreMgd')
clr.AddReference('AcDbMgd')
clr.AddReference('AecBaseMgd')
clr.AddReference('AecPropDataMgd')
clr.AddReference('AeccDbMgd')

# Import references from AutoCAD
from Autodesk.AutoCAD.Runtime import *
from Autodesk.AutoCAD.ApplicationServices import *
from Autodesk.AutoCAD.EditorInput import *
from Autodesk.AutoCAD.DatabaseServices import *
from Autodesk.AutoCAD.Geometry import *

# Import references from Civil3D
from Autodesk.Civil.ApplicationServices import *
from Autodesk.Civil.DatabaseServices import *

# The inputs to this node will be stored as a list in the IN variables.
pressure_pipes = IN[0]
primary_surface = IN[1]
secondary_surface = IN[2]

adoc = Application.DocumentManager.MdiActiveDocument
editor = adoc.Editor

debug = []

# Function to get elevation at a given point on the surface
def get_elevation_at_point(surface, point):
    try:
        elevation = surface.FindElevationAtXY(point.X, point.Y)
        return elevation
    except Exception as e:
        # Catch all exceptions and log them, allowing the script to continue
        debug.append(f"Error at point ({point.X}, {point.Y}): {str(e)}")
        return None

def set_pressure_pipe_surface(pipes, primary, secondary):
    """Set alignment reference for each pressure pipe if applicable."""
    with adoc.LockDocument():
        with adoc.Database.TransactionManager.StartTransaction() as t:
            primary = t.GetObject(primary.InternalObjectId, OpenMode.ForRead)
            secondary = t.GetObject(secondary.InternalObjectId, OpenMode.ForRead)
            for pipe in pipes:
                try:
                    pipe = t.GetObject(pipe.ObjectId, OpenMode.ForWrite)
                    start = pipe.StartPoint
                    end = pipe.EndPoint
                    start_elev = get_elevation_at_point(primary, start)
                    if start_elev:
                        end_elev = get_elevation_at_point(primary, end)
                        if end_elev:
                            pipe.ReferenceSurfaceId = primary.ObjectId
                    else:
                        start_elev = get_elevation_at_point(secondary, start)
                        if start_elev:
                            end_elev = get_elevation_at_point(secondary, end)
                            if end_elev:
                                pipe.ReferenceSurfaceId = secondary.ObjectId
                    if not start_elev:
                        debug.append(f"No surface found for pipe {pipe.Name}")
                except Exception as e:
                    # Log any errors that occur while processing each pipe
                    debug.append(f"Error processing pipe {pipe.Name}: {str(e)}")
            t.Commit()
            return primary

# Execute the function
output = set_pressure_pipe_surface(pressure_pipes, primary_surface, secondary_surface)

# Assign your output to the OUT variable.
OUT = output

Best to use the preformatted text option when sharing code. Edited to help those who might want to copy/edit more readily.

On mobile so I can’t get the desktop view, but this screenshot should help you find what you need (red than blue).

It’s best practice to only open objects when you modify them, also is the pipe/primary a Dynamo object or a C3D object? maybe use InternalObjectId. Add a continue after updating the pipe

pipe = t.GetObject(pipe.InternalObjectId, OpenMode.ForRead)
# ...
if end_elev:
    pipe.UpgradeOpen()
    pipe.ReferenceSurfaceId = primary.ObjectId
    continue  # Go to next pipe as no need to process further

You could simplify the logic as you are not using the output of the get_elevation_at_point function and rather than having multiple try statements - if this throws an exception it can be captured by the main function

def pipe_on_surface(surface, pipe):
    """Check if pipe is aligned to the surface xy coordinate space"""
    return all(
        [
            surface.FindElevationAtXY(point.X, point.Y)
            for point in [pipe.StartPoint, pipe.EndPoint]
        ]
    )

Edit: This will return false if both elevations are zero

Thank you for the help. I am new around here obviously.

1 Like

I tried implementing these changes, but I am still running into the same exception that is stopping the rest of my code from running. Any other suggestions, or maybe they were implemented incorrectly? I have uploaded the whole script and the revised python code.

I also made a small change to how I am getting the surfaces since I have been running into an issue with the Autodesk package surface selection node. The surfaces and the pipes should both be DatabaseServices objects.

Set Refence Surface (24.08.21).dyn (19.4 KB)

# Load the Python Standard and DesignScript Libraries
import sys
import clr

# Add Assemblies for AutoCAD and Civil3D
clr.AddReference('AcMgd')
clr.AddReference('AcCoreMgd')
clr.AddReference('AcDbMgd')
clr.AddReference('AecBaseMgd')
clr.AddReference('AecPropDataMgd')
clr.AddReference('AeccDbMgd')

# Import references from AutoCAD
from Autodesk.AutoCAD.Runtime import *
from Autodesk.AutoCAD.ApplicationServices import *
from Autodesk.AutoCAD.EditorInput import *
from Autodesk.AutoCAD.DatabaseServices import *
from Autodesk.AutoCAD.Geometry import *

# Import references from Civil3D
from Autodesk.Civil.ApplicationServices import *
from Autodesk.Civil.DatabaseServices import *

# The inputs to this node will be stored as a list in the IN variables.
pressure_pipes = IN[0]
primary_surface_name = IN[1]
secondary_surface_name = IN[2]

adoc = Application.DocumentManager.MdiActiveDocument
editor = adoc.Editor

debug = []

# Function to find the TinSurface by name
def find_surface_by_name(surface_name, t):
    try:
        civil_db = CivilApplication.ActiveDocument
        surfaces = civil_db.GetSurfaceIds()
        for surface_id in surfaces:
            surface_obj = t.GetObject(surface_id, OpenMode.ForRead)
            if surface_obj.Name == surface_name:
                return surface_obj
        return None
    except Exception as e:
        debug.append(f"Error finding surface by name: {str(e)}")
        return None

def pipe_on_surface(surface, pipe):
    """Check if pipe is aligned to the surface xy coordinate space"""
    try:
        return all(
            surface.FindElevationAtXY(point.X, point.Y) is not None
            for point in [pipe.StartPoint, pipe.EndPoint]
        )
    except Exception as e:
        debug.append(f"Error checking pipe alignment: {str(e)}")
        return False

def set_pressure_pipe_surface(pipes, primary_name, secondary_name):
    """Set alignment reference for each pressure pipe if applicable."""
    with adoc.LockDocument():
        with adoc.Database.TransactionManager.StartTransaction() as t:
            primary = find_surface_by_name(primary_name, t)
            secondary = find_surface_by_name(secondary_name, t)
            
            for pipe in pipes:
                try:
                    pipe = t.GetObject(pipe.ObjectId, OpenMode.ForRead)
                    if primary and pipe_on_surface(primary, pipe):
                        pipe.UpgradeOpen()  # Needed to modify the pipe
                        pipe.ReferenceSurfaceId = primary.ObjectId
                        continue
                    elif secondary and pipe_on_surface(secondary, pipe):
                        pipe.UpgradeOpen()  # Needed to modify the pipe
                        pipe.ReferenceSurfaceId = secondary.ObjectId
                        continue
                    else:
                        debug.append(f"No valid surface found for pipe {pipe.Name}")
                except Exception as e:
                    debug.append(f"Error processing pipe {pipe.Name}: {str(e)}")
            t.Commit()
            return None

# Execute the function
output = set_pressure_pipe_surface(pressure_pipes, primary_surface_name, secondary_surface_name)

# Assign your output to the OUT variable.
OUT = output

pipe.ReferenceSurfaceId property is incorrect - it is pipe.RefSurfaceId see API docs here

Hey Mike, thank you for the help. I believe it should be ReferenceSurfaceId since this is for pressure pipes. I apologize if I didn’t make that clear. I was trying to keep the function variables shorter to just pipes. You can see that API reference Here

Your link was for a Civil 3D part object (the parent of a pipe object) which also uses the RefSurfaceId property
The link for pressure pipes is here which does use ReferenceSurfaceId

If the code is still throwing exceptions I would try the following

  • Limit the use of try... except to code you know will throw an exception
  • Use above in conjunction with the python traceback library which can get more useful information from inside transactions
  • Unwrap the functions unless they are being used more than once and work with one transaction around the code
  • Work back by commenting out lines until the code works
  • If you are not using an IDE use that to help format the code and pick up type errors
  • Be persistent. C3D can be a fussy beast

Some example code to build ObjectId dictionaries of various database objects - you can then pull the id by using the name using dict.get() method as it returns None if it fails

import clr

clr.AddReference("AcDbMgd")
clr.AddReference("AeccDbMgd")

from Autodesk.AutoCAD.ApplicationServices import *
from Autodesk.AutoCAD.DatabaseServices import *

from Autodesk.Civil.ApplicationServices import *
from Autodesk.Civil.DatabaseServices import *


def build_dict(id_collection):
    return {id.GetObject(OpenMode.ForRead).Name: id for id in id_collection}


adoc = Application.DocumentManager.MdiActiveDocument
cdoc = CivilDocument.GetCivilDocument(adoc.Database)

with adoc.LockDocument():
    with adoc.Database as db:
        with db.TransactionManager.StartTransaction() as t:
            ppnIds = CivilDocumentPressurePipesExtension.GetPressurePipeNetworkIds(cdoc)
            ppn_dict = build_dict(ppnIds)
            pnIds = cdoc.GetPipeNetworkIds()
            pn_dict = build_dict(pnIds)
            sufaceIds = cdoc.GetSurfaceIds()
            surface_dict = build_dict(sufaceIds)

            t.Commit()

OUT = ppn_dict, pn_dict, surface_dict

image