返回顶部

在 Python 脚本中使用 web 工具

在本主题中

  1. 使用 ArcPy
  2. 使用 REST

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

注：

以下示例将使用联合 ArcGIS Server 上的地理处理服务。 您不会使用门户中相应的 web 工具项目。

使用 ArcPy

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

使用 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 工具的权限，请提供用户名和密码。

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

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

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

注：

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

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

相关主题

  • GUID-A393461A-1A4D-4D45-8914-3F1AFA57AB2B

有关此主题的反馈?

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