在 Python 脚本中使用地理处理服务—ArcGIS Pro

可以编写通过多种方法执行和使用地理处理服务的 Python 脚本。执行脚本的主要方法是使用 ArcPy。ArcPy 内置用来连接、执行和处理服务结果的方法。此外，如果访问 ArcGIS Server 服务目录中的服务，可以通过内置的 Python 模块调用使用 JSON 结构的 REST，以便传输结果。您将必须使用 Python 代码在临时工作空间构建客户端以便使用 ArcPy。大多数脚本均通过 ArcPy 来连接和使用地理处理服务。

注：

这里的示例将使用实际 ArcGIS Server 服务。不会使用门户的地理处理服务。

使用 ArcPy

在 ArcGIS Pro中，可以通过 Python 窗口（脚本工具或独立脚本）来访问地理处理服务。服务 URL 用于连接和使用地理处理服务。

使用 ImportToolbox 连接服务。


# arcpy.ImportToolbox("http://<hostname>:<port>/arcgis/services;<optional folder>/<service name>", "<optional alias>")
arcpy.ImportToolbox("https://degrassi/arcgis/services;GPFolder/BufferService", "PointAlias")

用于导入该工具的路径是地理处理服务的 URL。ImportToolbox 接受两个参数：该服务的 URL 和工具箱的可选别名。URL 参数分为使用分号 (;) 隔开的两部分。第一部分是服务端点 URL（或链接），第二部分是服务名称（或者，文件夹名称在服务名称之前）。

地理处理服务既可以同步执行，也可以异步执行。作为 Python 脚本编写者，您需要了解如何执行服务才能使用该服务。IsSynchronous 属性可用来确定服务的执行类型。当同步执行服务时，会自动返回结果，但在服务结束之前不能执行任何操作。当异步执行服务时，必须定期查询当前执行状态。完成服务执行后，可以访问结果。

下列 Python 代码显示了如何连接到异步地理处理服务以及如何使用 ArcPy 功能来执行服务、获取结果和进一步处理服务。通过在执行任务时设置结果变量，可使用 while 循环检查状态。返回状态码 4 （成功）或更高状态码时，表示任务完成。

使用异步服务创建缓冲区并在本地保存结果。

import arcpy import time
arcpy.ImportToolbox("http://sampleserver6.arcgisonline.com/ArcGIS/services;Elevation/ESRI_Elevation_World", "viewshedAlias")
result = arcpy.Viewshed_viewshedAlias(r'c:\data\inputPoint.shp', "10000 Kilometers")
while result.status < 4:
	   print(result.status)	   time.sleep(0.2) print("Execution Finished")
print(result.getMessages())
arcpy.CopyFeatures_management(result[0], 'localResult.shp')

有关 Result 对象和状态码及创建栅格和地图图像输出的详细信息，请参阅帮助主题在 Python 中使用工具。

使用 REST

使用地理处理服务的另一个方法（但不常用）是通过编写用于执行 REST 调用的脚本，同时使用 JSON 作为数据交换格式。此方法要求编写发送请求和处理响应的代码。

由于必须自行处理所有输入和输出语法，因此 REST 消息的发送和接收较为复杂。当发送和接收 REST 消息时，这些消息能够以一致方式返回。请求可以通过 HTTP GET 或 HTTP POST 方法发送，而响应则以 JSON 结构返回。核心 Python 库同时支持发送请求和函数，这更易于读取和解析 JSON 消息。

下例使用 Esri 示例服务器上的服务，并说明如何连接服务、发送请求和处理响应。

提交请求

例如，SampleServer6 包含用于创建视域的地理处理服务，并且可以从 http://sampleserver6.arcgisonline.com/ArcGIS/rest/services/Elevation 中访问。此服务采用点和距离作为输入，并返回要素集。下列输入有助于更好地了解该服务：


参数名称	输入值
输入观测点 (GPFeatureRecordSetLayer)	{"geometryType" : "esriGeometryPoint", "spatialReference" : {"wkid" : 54003}, 'features':[{'geometry':{'x': -13308192.1956127, 'y': 4221903.58555983}}]}
视域距离 (GPLinearUnit)	{ 'distance' : 8.5, 'units' : 'esriMiles' }
格式（输出格式）	JSON

此请求从服务中返回可见位置的 JSON。用于生成该请求的完整 URL 可在 Web 浏览器地址栏中使用。

http://sampleserver6.arcgisonline.com/ArcGIS/rest/services/Elevation/ESRI_Elevation_World/GPServer/Viewshed/execute?Input_Observation_Point={%22features%22%3A[{%22geometry%22%3A{%22x%22%3A-13308192.1956127%2C%22y%22%3A4221903.58555983}}]}&Viewshed_Distance={+%27distance%27+%3A+8.5%2C+%27units%27+%3A+%27esriMiles%27+}&env%3AoutSR=&env%3AprocessSR=&f=pjson

使用 Python 模块 urllib 或 urllib2 可以提交 URL。下列 Python 代码将以使用服务目录或将链接复制到 Web 浏览器地址栏的类似方法来执行上述请求。

# Requires Python 3+
import urllib.request as urlopen import urllib.parse as urlencode import urllib.request as request import json
inPts = {"geometryType" : "esriGeometryPoint",         "spatialReference" : {"wkid" : 54003},         'features':[{'geometry': {'x': -13308192.1956127, 'y': 4221903.58555983}}]}
dist = {'distance':8.5,'units':'esriMiles'}
data = {'Input_Observation_Point': inPts,        'Viewshed_Distance': dist,        'f': 'pjson'}
URL = 'http://sampleserver6.arcgisonline.com/ArcGIS/rest/services/Elevation/ESRI_Elevation_World/GPServer/Viewshed/execute'
req = request.Request(URL, urlencode.urlencode(data).encode('UTF-8')) response = urlopen.urlopen(req) response_bytes = response.read()
print(json.loads(response_bytes.decode('UTF-8')))

Result 对象以字符串格式返回。共有多种生成请求和解析结果的方法；上例仅是其中一种方法。使用 json.loads() 方法可以将地理处理服务中返回的 JSON 放到字典中，如上例所示。根据地理处理服务的输出，这可能是最佳技术，否则，在 Python 中通过 REST 使用地理处理服务时，可能需要浏览其他选项来处理输出。

注：

如果使用结果要素，则确保使用实际 x,y 坐标对对工作流有意义，因为您将无法以支持 ArcGIS 内几何的格式接收实际要素。

异步执行的服务要求定期请求任务状态，以便查看是否完成任务，这与上述使用 ArcPy 的示例类似。Python 代码可以在检查作业状态时返回的 jobStatus 状态消息中查找 esriJobSucceeded 或 esriJobFailed 短语。下列代码示例表示使用异步地理处理服务（使用 submitJob）的一项技术：

import urllib.request as urlopen import urllib.parse as urlencode import urllib.request as request import json import time
def sendReq(URL, data=None):
    req = request.Request(URL, urlencode.urlencode(data).encode('UTF-8'))    response = urlopen.urlopen(req)    response_bytes = response.read()    return json.loads(response_bytes.decode('UTF-8'))
inPts = {"geometryType": "esriGeometryPoint",         "spatialReference": {"wkid" : 54003},         'features': [{'geometry': {'x': -13308192.1956127, 'y': 4221903.58555983}}]}
dist = {'distance': 8.5, 'units': 'esriMiles'}
data = {'Input_Observation_Point': inPts,        'Viewshed_Distance': dist,        'f': 'pjson'}
taskUrl = "http://localhost:6080/arcgis/rest/services/WorldViewshed/GPServer/viewshed"
submitUrl = taskUrl + "/submitJob"
submitJson = sendReq(submitUrl, data)
if 'jobId' in submitJson:
    jobID = submitJson['jobId']    status = submitJson['jobStatus']    jobUrl = taskUrl + "/jobs/" + jobID
    while status == "esriJobSubmitted" or status == "esriJobExecuting":
        print("checking to see if job is completed...")        time.sleep(1)
        jobJson = sendReq(jobUrl, {"f": "json"})
        if 'jobStatus' in jobJson.keys():
            status = jobJson['jobStatus']
            if status == "esriJobSucceeded":
                if 'results' in jobJson:
                    resultsJson = jobJson['results']                    for paramName in resultsJson.keys():
                        resultsUrl = jobUrl + "/" + resultsJson[paramName]['paramUrl']                        resultJson = sendReq(resultsUrl, {"f": "json"})                        print(resultJson)
            if status == "esriJobFailed":
                if 'messages' in jobJson.keys():
                    print(jobJson['messages'])
else:
    print("no jobId found in the response")

有关此主题的反馈?