Restful API 示例代码
这些示例代码用于通过应用程序编程接口 (Application Programming Interfaces, API) 从 FactoryTalk 系统状态门户获取信息。所有代码样本基于软件版本 3.7 开发。
import requests
import json
import urllib3
import base64
import time
from dateutil.relativedelta import relativedelta
from datetime import datetime
urllib3.disable_warnings() # 用于禁用 ssl 认证。
常量
scope_ads_path = "RNA://$Global/"
api_base = "FTSystemStatus/"
api_resource_alarm = "alarm/"
api_prefix_hmi = "hmi/"
api_prefix = "api/"
api_v1 = "v1/"
api_resource_auth = "auth/"
api_prefix_alarm = api_base + api_prefix + api_v1 + api_resource_alarm
api_auth_login = "login"
api_auth_authentication = "authenticate"
api_applications = "applications"
api_servers = "servers"
api_hmi_connections = "connections"
api_hmi_client_connections = "client-connections"
api_alarm_db_name = "dbname/"
api_alarm_summary = "summary/"
api_alarm_top_frequency = "top-frequencies/"
api_alarm_priority_distribution = "priority-distributions/"
api_alarm_top_stale_alarms = "top-staleness/"
api_alarm_occurrences = "occurrences/"
api_auth_logout = "logout"
default_https_port = "443"
default_http_port = "80"
server_ip = "xxx.xxx.xxx.xxx"
server_port = "443"
class Type(enumerate):
HMIServer = 0
AlarmServer = 1
DataServer = 2
OPCDAServer = 3
OPCUAServer = 4
HistorianSE = 5
HistorianME = 6
功能
def Login(server_address, username, password):
""" 此方法用于登录 FactoryTalk 系统状态门户。
它将用作登录 FactoryTalk 系统状态门户的身份验证。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
username：FactoryTalk 的用户名
password：FactoryTalk 的密码
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_resource_auth + api_auth_login
requester = json.dumps({
"userName": username,
"password": password
})
response = requests.request("POST", url, headers=GetRequestHeaders("", ""), data=requester, verify=False)
response_content = response.content.decode()
login_response = json.loads(response_content)['authentication']
print("login response:")
print(login_response)
print("")
if server_port == ( default_https_port or default_http_port ):
server_cookie_name = server_ip
else:
server_cookie_name = server_ip + ":"+ server_port
if 'localhost' in server_cookie_name:
server_cookie_name = 'localhost.local'
cookie = response.cookies._cookies[server_cookie_name]['/']['ASP.NET_SessionId']
session_id = cookie.value
token = json.loads(response_content)['authentication']['accessToken']
return session_id, token
def Authentication(server_address, session_id):
""" 此方法用于验证当前会话是否已登录。
它将用作登录 FactoryTalk 系统状态门户的身份验证。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
session_id：从登录界面标题返回
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_resource_auth + api_auth_authentication
response = requests.request("GET", url, headers=GetRequestHeaders("", session_id), verify=False)
response_content = response.content.decode()
authentication_response = json.loads(response_content)['authentication']
print("authentication response:")
print(authentication_response)
print("")
def Applications(server_address, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取应用程序列表。
它将用作登录 FactoryTalk 系统状态门户的身份验证。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'fullName': "user",
'isAbnormal': true,
'applicationState':0# response get three states:0:NotSet,1:IsNormal,2:IsAbnormal
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_applications
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
app_list = json.loads(response_content)['applicationList']
print("GetApplications response:")
print(app_list)
print("")
return app_list
def Servers(server_address, application_name, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取服务器列表。
用于从 FactoryTalk 系统状态门户获取您在 GetApplication 界面中
选择的应用程序下所有受支持的服务器列表。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
application_name：GetApplications 界面中的应用程序名称。
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'applicationName': "WebPortalAlarm",
'serverList':[{
'fullName':"RNA://$Global/WebPortalAlarm/Area1/FTAE1",
'state':1, #0:Running,1:Loading,2:Exception
'priComputer':{
'name':"DESKTOP-49RNNEM",
'state':4 #0:Active,1:Standby,2:Notloaded,3:Failed,4:Transition,5:Unknown
}
'type':1 #0:HMIServer,1:AlarmServer,2:DataServer,3:OPCDAServer,4:OPCUAServer,5:HistorianSE,6:HistorianME
}]
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
application_name_encode = Encode(application_name)
url = server_address + api_base + api_prefix + api_v1 + api_servers + "/" + application_name_encode
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
server_list = json.loads(response_content)['appServer']['serverList']
print("GetServers response:")
print(server_list)
print("")
return server_list
def Connections(server_address, server_name, host_name, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取 HMI 连接列表。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 HMI 服务器名称
host_name：HMI 服务器计算机名称
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
connectionList:[{
'serverName': "RNA://$Global/ClientTagApp1/Area 1/ClientTag1HMI1",
'hostName': "WEBPORTALFTVP",
'clientIP': "10.***.***.***",
'clientName':"DESKTOP-SS-1",
'connectMessage':"2/28/2023 6:54:11 AM - 3/6/2023 7:38:13 AM",
'connectDuration':"0.0",
'connectState':False,
'connectionDurationSeconds': 521042.0
}]
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_prefix_hmi + api_hmi_connections + "/" + server_name + "/" + host_name
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
hmi_connections_response = json.loads(response_content)['connectionList']
print("GetHMIConnections response:")
print(hmi_connections_response)
print("")
return hmi_connections_response
def ClientConnections(server_address, server_name, host_name, client_name, headers):
""" 此方法用于获取 HMI 连接历史列表。
参数：
server_address：请求服务器地址
server_name：目标 HMI 服务器名称
host_name：HMI 服务器计算机名称
client_name：FT View SE Client 计算机名称
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
connectionList:[{
'serverName': "ClientTag1HMI1",
'ServerPCName': "WEBPORTALFTVP",
'ServerIPAddress': " ",
'UserName':"DESKTOP-SS-1\\ADMINISTRATOR",
'StartTimeUTC':"2/28/2023 6:54:11 AM",
'EndTimeUTC':"--",
'PID':4572
}]
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_prefix_hmi + api_hmi_client_connections + "/" + server_name + "/" + host_name + "/" + client_name
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
hmi_historical_response = json.loads(response_content)['connectionList']
print("GetHMIHistoricalConnections response:")
print(hmi_historical_response)
print("")
def DBName(server_address, server_name, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取报警数据库名称。
用于获取所选应用程序下所有受支持的服务器列表
从 FactoryTalk 系统状态门户 GetApplication 接口。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 FactoryTalk Alarms and Events、OPC UA 或 OPC DA 服务器名称
headers：请求标题
返回值：
将返回字符串类型数据。
例如：
'WebPortalAlarm'
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_db_name + server_name
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
print("GetDBName response:")
print(response_content)
print("")
def AlarmSummary(server_address, server_name, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取报警汇总。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 FactoryTalk Alarms and Events、OPC UA 或 OPC DA 服务器名称
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'CountOfTotal':2,
'CountOfInAlarm':0,
'CountOfUnacked':0,
'CountOfShelved':0,
'CountOfSuppressed':0,
'CountOfDisabled':0,
'CountOfBad':2
}
引发：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_summary + server_name
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
alarm_count_response = json.loads(response_content)['Data']
print("GetAlarmSummary response:")
print(alarm_count_response)
print("")
def TopFrequency(server_address, server_name, begin_time, end_time, headers):
""" 此方法用于从 FactoryTalk 系统状态门户获取一段时间内状态频繁更改的前 10 个报警的数量。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 HMI 服务器名称
begin_time：开始时间
end_time：结束时间
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'Items':[{
'Proportion':0.8079424,
'Name':"WebPortalAlarm/Area1/FTAE1/Alarm1 LOLO",
'Quantity':33305
}]
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_top_frequency + server_name + "/" + begin_time + "/" + end_time
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
alarm_top_frequent_response = json.loads(response_content)['Data']
print("GetTopFrequentAlarms response:")
print(alarm_top_frequent_response)
print("")
def PriorityDistribution(server_address, server_name, begin_time, end_time, headers):
""" 此方法用于获取一段时间内各优先级的报警数量和比率。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 FactoryTalk Alarms and Events、OPC UA 或 OPC DA 服务器名称
begin_time：开始时间
end_time：结束时间
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'Items':[{
'Priority':4,
'Quantity':18132,
'Proportion':0.996373236
}]
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_priority_distribution \
+ server_name + "/" + begin_time + "/" + end_time
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
alarm_priority_distribution_response = json.loads(response_content)['Data']
print("GetPriorityDistribution response:")
print(alarm_priority_distribution_response)
print("")
def TopStaleness(server_address, server_name, begin_time, end_time, headers):
""" 此方法用于获取一个时间段内活动状态持续时间最长的前 10 个报警。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 FactoryTalk Alarms and Events、OPC UA 或 OPC DA 服务器名称
begin_time：开始时间
end_time：结束时间
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
例如：
{
'Items':[{
'AlarmName':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Stale':164487,
}]
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_top_stale_alarms + server_name + "/" + begin_time + "/" + end_time
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
alarm_top_stale_response = json.loads(response_content)['Data']
print("GetTopStaleAlarms response:")
print(alarm_top_stale_response)
print("")
def Occurrences(server_address, server_name, begin_time, end_time, headers):
""" 此方法用于获取一段时间内报警的数量和优先级分配。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
server_name：目标 FactoryTalk Alarms and Events、OPC UA 或 OPC DA 服务器名称
begin_time：开始时间，如果该时间小于一小时，则按 1 小时计算
end_time：结束时间
headers：请求标题
返回值：
它将返回一组 JSON 格式的数据。
每组数据都包含多个参数
例如：
{
'Items':[{
'Time':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Urgent':164487,
'High':0,
'Medium':0,
'Low':0,
'UtcTime':"1678060800000"
}]
}
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_prefix_alarm + api_alarm_occurrences + server_name + "/" + begin_time + "/" + end_time
response = requests.request("GET", url, headers=headers, verify=False)
response_content = response.content.decode()
alarm_occurrences_response = json.loads(response_content)['Data']
print("GetAlarmOccurrences response:")
print(alarm_occurrences_response)
print("")
def LogOut(server_address, session_id):
""" 此方法用于注销 FactoryTalk 系统状态门户。
参数：
server_address：请求 FactoryTalk 系统状态门户服务器地址
session_id：从登录界面标题返回
返回值：
引发：
我们使用标准 HTTP 网络协议，响应代码是返回的标准状态代码。
例如：
响应代码 500 是内部网络错误。
响应代码 401 未经授权用于此服务。
"""
pass
url = server_address + api_base + api_prefix + api_v1 + api_resource_auth + api_auth_logout
payload = ""
response = requests.request("POST", url, data=payload, headers=GetRequestHeaders("", session_id), verify=False)
response_content = response.content.decode()
response_code = json.loads(response_content)['responseCode']
if response_code == 200:
print("logout successfully.")
def GetRequestHeaders(token, session_id):
headers = {
'Content-Type': "application/json",
'Authorization': "Bearer " + token,
'Host': server_ip + ':'+ server_port,
'Cookie': "ASP.NET_SessionId=" + session_id,
}
return headers
def Encode(data):
encode_data = str(base64.b64encode(data.encode()), encoding='utf-8')
return encode_data
def main():
username = "xxx" # FactoryTalk's username
password = "xxx" # FactoryTalk's password
server_address = "https://"+ server_ip + ":"+ server_port + "/"# FT 系统状态 Web 服务器地址
# 稍后需要时从登录方法获取令牌和会话 ID
response_authentication_message = Login(server_address, username, password)
session_id = response_authentication_message[0]
token = response_authentication_message[1]
# 此方法用于前端开发验证
# 验证（server_address、session_id）
# 包含访问令牌和会话 ID 的请求标头
headers = GetRequestHeaders(token, session_id)
# 获取当前应用程序列表
response_application_list = Applications(server_address, headers)
target_hmi_application_name = "ClientTagApp1" # 要获取的演示应用程序名称
target_ftae_application_name = "WebPortalAlarm_History"
for i in range(len(response_application_list)):
if response_application_list[i]['fullName'] == target_hmi_application_name:
app_hmi_name = response_application_list[i]['fullName']
elif response_application_list[i]['fullName'] == target_ftae_application_name:
app_ftae_name = response_application_list[i]['fullName']
# 获取当前服务器列表
response_hmi_server_list = Servers(server_address, app_hmi_name, headers)
response_ftae_server_list = Servers(server_address, app_ftae_name, headers)
# target_server_name 是要获取的演示服务器名称
hmi_area = "/Area 1"
hmi_name = "/ClientTag1HMI1"
target_hmi_server = scope_ads_path + app_hmi_name + hmi_area + hmi_name
# 应用程序根目录下的目标 FTAE 服务器
ftae_area = "/Area1"
ftae_name = "/FTAE3"
target_ftae_server = scope_ads_path + app_ftae_name + ftae_area + ftae_name
for i in range(len(response_hmi_server_list)):
if response_hmi_server_list[i]['fullName'] == target_hmi_server \
and response_hmi_server_list[i]['type'] == Type.HMIServer:
host_name = response_hmi_server_list[i]['priComputer']['name']
hmi_server_name = response_hmi_server_list[i]['fullName']
break
for j in range(len(response_ftae_server_list)):
if response_ftae_server_list[j]['fullName'] == target_ftae_server:
ftae_server_name = response_ftae_server_list[j]['fullName']
break
host_name_encode = Encode(host_name)
hmi_server_name_encode = Encode(hmi_server_name)
# 此接口用于获取当前连接的 HMI 信息
Connections(server_address, hmi_server_name_encode, host_name_encode, headers)
# 连接到 HMI 服务器的 HMI 客户端
client_name = 'DESKTOP-SS-1'
client_name_encode = Encode(client_name)
# 此接口用于获取当前连接的 HMI 历史信息
ClientConnections(server_address, hmi_server_name_encode, host_name_encode, client_name_encode, headers)
ftae_server_name_encode = Encode(ftae_server_name)
# 获取 FTAE 数据库信息
DBName(server_address, ftae_server_name_encode, headers)
# 获取报警汇总
AlarmSummary(server_address, ftae_server_name_encode, headers)
# 将时间跨度设置为上周
begin_time = str(int(round(datetime.timestamp(datetime.now() + relativedelta(weeks=-2)) * 1000)))
# 这是当前时间
end_time = str(int(round(time.time() * 1000)))
# 用于查询报警的状态变化数。
TopFrequency(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# 用于获取设置时间内每个级别的报警触发器数
PriorityDistribution(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# 用于获取某一时间段内前 10 个活动时间最长的报警
TopStaleness(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# 用于获取已触发报警的数量
Occurrences(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# 用于注销
LogOut(server_address, session_id)
if __name__ == '__main__':
main()
提供反馈