I am relatively new to Python and thought I would give a stab at multiprocessing. I have a script that runs well in IDLE or as an ArcMap toolbox script. After perusing these forums and docs.python, I made an attempt at incorporating my working script into a multiprocessing one. However similar working examples on this forum are, none address data processing as I would like to. I hope it is feasible.
Basically, the script moves through a list of elevation rasters (ERDAS IMG format), extracting cells below a threshold, and finally merging them together. I am currently running the script in the command prompt, as everything else opens new windows, or crashes in an attempt to. The script gives the illusion that it works fine, except it seems to move on to the final merge before waiting for the workers to completely finish.
I have looked at several examples and few seem to have more than a couple of processes in the worker function. None of which are arcpy geoprocesses. So I guess my questions are essentially
Should I be using something other than pool.apply_async, such as pool.map or pool.apply?
Am I properly returning the path of the final polygon to the resultList?
Import modules
import arcpy, os, math from arcpy import env from arcpy.sa import * import multiprocessing import time
Check out licenses
arcpy.CheckOutExtension("spatial")
Define functions
def worker_bee(inputRaster, scratch, addNum):
(path, lName) = os.path.split(inputRaster) (sName, ext) = os.path.splitext(lName) nameParts = sName.split("_") nameNumber = nameParts[-1]# Create scratch subfolder if not exists subFolder = scratch + "\\" + nameNumber + "_output" if not os.path.exists(subFolder):os.makedirs(subFolder) # Set workspace to subfolder arcpy.env.workspace = subFolder arcpy.env.overwriteOutput=True arcpy.env.extent = "MAXOF" # Local Variables Expression = "Shape_Area >= 100" poly1 = subFolder + "\\poly1.shp" poly2 = subFolder + "\\poly2.shp" poly3 = subFolder + "\\poly3.shp" poly4 = subFolder + "\\poly4.shp" poly5 = subFolder + "\\poly5.shp" poly6 = subFolder + "\\poly6.shp" poly7 = subFolder + "\\poly7.shp" outName = scratch + "\\ABL_" + nameNumber + ".shp" #### Perform calculations ### # Map Algebra (replace -9999 with 9999) inRasterCon = Con(inputRaster, 9999, inputRaster, "Value = -9999") # Filter DEM to smooth out low outliers filterOut = Filter(inRasterCon, "LOW", "DATA") # Determine raster MINIMUM value and calculate threshold filterMinResult = arcpy.GetRasterProperties_management(filterOut, "MINIMUM") filterMin = filterMinResult.getOutput(0) threshold = (float(filterMin) + float(addNum)) # Map Algebra (values under threshold) outCon = Con(filterOut <= threshold, 1, "") arcpy.RasterToPolygon_conversion(outCon, poly1, "SIMPLIFY", "Value") # Dissolve parts arcpy.Dissolve_management(poly1, poly2, "", "", "SINGLE_PART", "DISSOLVE_LINES") # Select parts larger than 100 sq m arcpy.Select_analysis(poly2, poly3, Expression) # Process: Eliminate Polygon Part arcpy.EliminatePolygonPart_management(poly4, poly5, "PERCENT", "0 SquareMeters", "10", "CONTAINED_ONLY") # Select parts larget than 100 sq m arcpy.Select_analysis(poly5, poly6, Expression) # Simplify Polygon arcpy.SimplifyPolygon_cartography(poly6, poly7, "BEND_SIMPLIFY", "3 Meters", "3000 SquareMeters", "RESOLVE_ERRORS", "KEEP_COLLAPSED_POINTS") # Smooth Polygon outShape = arcpy.SmoothPolygon_cartography(poly7, outName, "PAEK", "3 Meters", "FIXED_ENDPOINT", "FLAG_ERRORS").getOutput(0) ### Calculations complete ### # Delete scratch subfolder arcpy.Delete_management(subFolder) print("Completed " + outShape + "...") return outShaperesultList = [] def log_result(result): resultList.append(result)
if name == "main": arcpy.env.overwriteOutput=True
# Read in parameters inFolder = raw_input("Input Folder: ")#arcpy.GetParameterAsText(0) addElev = raw_input("Number of elevation units to add to minimum: ") # Create scratch folder workspace scratchFolder = inFolder + "\\scratch" if not os.path.exists(scratchFolder):os.makedirs(scratchFolder) # Local variables dec_num = str(float(addElev) - int(float(addElev)))[1:] outNameNum = dec_num.replace(".", "") outMerge = inFolder + "\\ABL_" + outNameNum + ".shp" # Print core usage cores = multiprocessing.cpu_count() print("Using " + str(cores) + " cores...") #Start timing start = time.clock() # List input tiles arcpy.env.workspace = inFolder inTiles = arcpy.ListRasters("*", "IMG") tileList = [] for tile in inTiles: tileList.append(inFolder + "\\" + tile) # Create a Pool of subprocesses pool = multiprocessing.Pool(cores) print("Adding jobs to multiprocessing pool...") for tile in tileList: # Add the job to the multiprocessing pool asynchronously pool.apply_async(worker_bee, (tile, scratchFolder, addElev), callback = log_result) # Clean up worker pool; waits for all jobs to finish pool.close() pool.join() # Get the resulting outputs (paths to successfully computed breakline polygons) #print("Getting resulting outputs...") #results = [job.get() for job in jobs] # Merge the temporary outputs print("Merging temporary outputs into shapefile " + outMerge + "...") arcpy.Merge_management(resultList, outMerge) # Clean up temporary data print("Deleting temporary data ...") for result in results: try: arcpy.Delete_management(result) except: pass # Stop timing and report duration end = time.clock() duration = end - start hours, remainder = divmod(duration, 3600) minutes, seconds = divmod(remainder, 60) print("Completed in %d:%d:%f" % (hours, minutes, seconds))
