Restful APIs' sample codes
These sample codes are used for getting the information from FactoryTalk System Status Portal with the Application Programming Interfaces (API). All code samples are developed on Python version 3.7.
import requests
import json
import urllib3
import base64
import time
from dateutil.relativedelta import relativedelta
from datetime import datetime
urllib3.disable_warnings() # It used to disable the ssl certification.
Constants
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
Functions
def Login(server_address, username, password):
""" This method is used to log in to FactoryTalk System Status Portal.
It will serve as your authentication for logging in to FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
username: FactoryTalk's username
password: FactoryTalk's password
Returns:
It will return a set of data in JSON format.
For example:
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used for verifying whether the current session is signed in.
It will serve as your authentication for logging in to the FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
session_id: Return from login interface headers
Returns:
It will return a set of data in JSON format.
For example:
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the application list from FactoryTalk System Status Portal.
It will serve as your authentication for logging in to FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'fullName': "user",
'isAbnormal': true,
'applicationState':0# response get three states:0:NotSet,1:IsNormal,2:IsAbnormal
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the server list from FactoryTalk System Status Portal.
It is used to obtain all supported server list under the application that you selected
in the GetApplication interface from FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
application_name: Application name from GetApplications interface.
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'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
}]
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the HMI connection list from FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target hmi server name
host_name: HMI server Computer name
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
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
}]
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the HMI connections historical list.
Args:
server_address: Request server address
server_name: Target hmi server name
host_name: HMI server computer name
client_name: FT ViewSE client computer name
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
connectionList:[{
'serverName': "ClientTag1HMI1",
'ServerPCName': "WEBPORTALFTVP",
'ServerIPAddress': " ",
'UserName':"DESKTOP-SS-1\\ADMINISTRATOR",
'StartTimeUTC':"2/28/2023 6:54:11 AM",
'EndTimeUTC':"--",
'PID':4572
}]
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the alarm database name from FactoryTalk System Status Portal.
It is used to obtain all supported server list under the application that you selected
in the GetApplication interface from FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target FactoryTalk Alarms and Events , OPC UA, or OPC DA server name
headers: Request headers
Returns:
It will return a string type data.
For example:
'WebPortalAlarm'
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the alarm summary from FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target FactoryTalk Alarms and Events , OPC UA, or OPC DA server name
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'CountOfTotal':2,
'CountOfInAlarm':0,
'CountOfUnacked':0,
'CountOfShelved':0,
'CountOfSuppressed':0,
'CountOfDisabled':0,
'CountOfBad':2
}
Raises:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the number of top 10 frequently status changed alarms in a period time from FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target hmi server name
begin_time: begin time
end_time: the end time
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'Items':[{
'Proportion':0.8079424,
'Name':"WebPortalAlarm/Area1/FTAE1/Alarm1 LOLO",
'Quantity':33305
}]
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the number and ratio of alarms of each priority in a period of time.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target FactoryTalk Alarms and Events , OPC UA, or OPC DA server name
begin_time: Begin time
end_time: End time
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'Items':[{
'Priority':4,
'Quantity':18132,
'Proportion':0.996373236
}]
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the top 10 alarms with the longest active state duration in a period time.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target FactoryTalk Alarms and Events , OPC UA, or OPC DA server name
begin_time: Begin time
end_time: End time
headers: Request headers
Returns:
It will return a set of data in JSON format.
For example:
{
'Items':[{
'AlarmName':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Stale':164487,
}]
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to get the number and priority distribution of alarms in a period time.
Args:
server_address: Request FactoryTalk System Status Portal server address
server_name: Target FactoryTalk Alarms and Events , OPC UA, or OPC DA server name
begin_time: Begin time, if the time less than one hour,it will be calculated as one hour
end_time: End time
headers: Request headers
Returns:
It will return a set of data in JSON format.
Each sets of data includes multiple parameters
For example:
{
'Items':[{
'Time':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Urgent':164487,
'High':0,
'Medium':0,
'Low':0,
'UtcTime':"1678060800000"
}]
}
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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):
""" This method is used to log out FactoryTalk System Status Portal.
Args:
server_address: Request FactoryTalk System Status Portal server address
session_id: Return from login interface headers
Returns:
Raises:
We use the standard HTTP network protocols, and the respond code is the standard status code returned.
For example:
The respondCode 500 is an internal network error.
The respondCode 401 is unauthorized for this service.
"""
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 system status web server address
# get token and session id we need after from login method
response_authentication_message = Login(server_address, username, password)
session_id = response_authentication_message[0]
token = response_authentication_message[1]
# This method is used for frontend develop authentication
# Authentication(server_address,session_id)
# request header with access token and session id
headers = GetRequestHeaders(token, session_id)
# get the current application list
response_application_list = Applications(server_address, headers)
target_hmi_application_name = "ClientTagApp1" # the demo application name you want to get
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']
# get the current server list
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 is the demo server name you want to get
hmi_area = "/Area 1"
hmi_name = "/ClientTag1HMI1"
target_hmi_server = scope_ads_path + app_hmi_name + hmi_area + hmi_name
# target ftae server under root of the application
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)
# # this interface is used to obtain the currently connected HMI information
Connections(server_address, hmi_server_name_encode, host_name_encode, headers)
# hmi client connected to hmi server
client_name = 'DESKTOP-SS-1'
client_name_encode = Encode(client_name)
# this interface is used to obtain the currently connected HMI historical information
ClientConnections(server_address, hmi_server_name_encode, host_name_encode, client_name_encode, headers)
ftae_server_name_encode = Encode(ftae_server_name)
# get FTAE database information
DBName(server_address, ftae_server_name_encode, headers)
# get alarm summary
AlarmSummary(server_address, ftae_server_name_encode, headers)
# set the time span as last week
begin_time = str(int(round(datetime.timestamp(datetime.now() + relativedelta(weeks=-2)) * 1000)))
# this is current time
end_time = str(int(round(time.time() * 1000)))
# used to query the number of status changes of the alarm.
TopFrequency(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# used to get the number of alarm triggers for each level within the set time
PriorityDistribution(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# used to get the top 10 alarms that have been active in a period time
TopStaleness(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# used to get the number of alarms triggered
Occurrences(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# used to log out
LogOut(server_address, session_id)
if __name__ == '__main__':
main()
Provide Feedback