返回顶部

在 Python 脚本中使用 web 工具

在本主题中

  1. 使用 ArcPy
  2. 使用 REST

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

注：

这里的示例将使用实际 ArcGIS Server 服务。不会使用门户的 web 工具。

使用 ArcPy

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

使用 ImportToolbox 连接服务。

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

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

web 工具既可以同步执行，也可以异步执行。作为 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

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

由于必须自行处理所有输入和输出语法，因此 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() 方法可以将 web 工具中返回的 JSON 放到字典中，如上例所示。根据 web 工具的输出，这可能是最佳方法，否则，在 Python 中通过 REST 使用 web 工具时，可能需要浏览其他选项来处理输出。

注：

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

异步执行的服务要求定期请求任务状态，以便查看是否完成任务，这与上述使用 ArcPy 的示例类似。Python 代码可以在检查作业状态时返回的 jobStatus 状态消息中查找 esriJobFailed 或 esriJobSucceeded 短语。下列代码示例演示使用异步 web 工具（使用 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")

相关主题

  • 地理处理 REST 服务概述
  • GUID-A393461A-1A4D-4D45-8914-3F1AFA57AB2B

有关此主题的反馈?

在本主题中
  1. 使用 ArcPy
  2. 使用 REST