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

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

使用 ArcPy

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

使用 ImportToolbox 或 AddToolbox 连接到服务。

# arcpy.ImportToolbox("https://machine.domain.com/webadaptor/services;<optional folder>/<service name>;{username};{password}")
arcpy.ImportToolbox("https://machine.domain.com/webadaptor/services;GPFolder/BufferService;serverusername;serverpassword")

用于导入该工具的路径是 web 工具的 URL。 ImportToolbox and AddToolbox 接受服务器的信息，其中包含以分号 (;) 分隔的 4 个部分。第一部分是服务端点 URL（或链接），第二部分是服务名称（可选，文件夹名称在服务名称之前），第三部分是可选的服务器用户名，最后一个部分是可选的服务器密码。如果您没有访问 web 工具的权限，请提供用户名和密码。

地理处理服务既可以同步运行，也可以异步运行。作为 Python 脚本编写者，您必须了解如何运行服务，才能使用该服务。 IsSynchronous 属性可用于确定服务是同步运行，还是异步运行。当服务同步运行时，将自动返回结果，但在服务完成之前，将无法执行任何操作。当服务异步运行时，必须定期查询当前状态。当服务完成运行后，即可访问结果。

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

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

import arcpy
import time

arcpy.ImportToolbox("http://machine.domain.com/webadaptor/services;Elevation/viewshedAlias;serverusername;serverpassword")

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 消息。

以下示例演示了如何连接到服务、发送请求和处理响应。

提交请求

例如，可以从 https://machine.domain.com/webadaptor/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 浏览器地址栏中使用。

https://machine.domain.com/webadaptor/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 = 'https://machine.domain.com/webadaptor/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 = "https://machine.domain.com/webadaptor/rest/services/Elevation/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")

有关此主题的反馈?