Exemples de code d'API Restful
Ces exemples de code sont utilisés pour obtenir des informations à partir du portail d'état du système FactoryTalk avec l'interface de programmation d'application (API, Application Programming Interface). Tous les exemples de code sont développés sur la version 3.7 de Python.
import requests
import json
import urllib3
import base64
import time
from dateutil.relativedelta import relativedelta
from datetime import datetime
urllib3.disable_warnings() # Utilisé pour désactiver la certification ssl.
Constantes
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
Fonctions
def Login(server_address, username, password):
""" Cette méthode est utilisée pour se connecter au portail d'état du système FactoryTalk.
Elle servira d'authentification pour vous connecter au portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
username : Nom d'utilisateur de FactoryTalk
password : Mot de passe de FactoryTalk
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour vérifier si la session en cours est connectée.
Elle servira d'authentification pour vous connecter au portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
session_id : Renvoyer à partir des en-têtes de l'interface de connexion
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'userName': "user",
'accessToken': "eyJhbGciOiJQUzI1NiIsInR5cCI6ImF0K2p3dCIsImtpZCI6InpOTGd3STR3TGtLOERGSjUxTzZjbWUtXzF"
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir la liste des applications à partir du portail d'état du système FactoryTalk.
Elle servira d'authentification pour vous connecter au portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'fullName': "user",
'isAbnormal': true,
'applicationState':0# response get three states:0:NotSet,1:IsNormal,2:IsAbnormal
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir la liste des serveurs à partir du portail d'état du système FactoryTalk.
Elle est utilisée pour obtenir toutes les listes de serveurs pris en charge sous l'application que vous avez sélectionnée
dans l'interface GetApplication du portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
application_name : Nom de l'application à partir de l'interface GetApplications.
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'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
}]
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir la liste des connexions IHM à partir du portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur IHM cible
host_name : Nom de l'ordinateur serveur IHM
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
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
}]
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir la liste historique des connexions IHM.
Args:
Server_address : Adresse du serveur de la demande
server_name : Nom du serveur IHM cible
host_name : Nom de l'ordinateur serveur IHM
client_name : Nom de l'ordinateur de FT View SE Client
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
connectionList:[{
'serverName': "ClientTag1HMI1",
'ServerPCName': "WEBPORTALFTVP",
'ServerIPAddress': " ",
'UserName':"DESKTOP-SS-1\\ADMINISTRATOR",
'StartTimeUTC':"2/28/2023 6:54:11 AM",
'EndTimeUTC':"--",
'PID':4572
}]
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir le nom de la base de données des alarmes à partir du portail d'état du système FactoryTalk.
Elle est utilisée pour obtenir toutes les listes de serveurs pris en charge sous l'application que vous avez sélectionnée
dans l'interface GetApplication du portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur cible FactoryTalk Alarms and Events, OPC UA ou OPC DA
headers : En-têtes de la demande
Renvoie :
Il renverra des données de type chaîne.
Par exemple :
'WebPortalAlarm'
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir le résumé des alarmes à partir du portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur cible FactoryTalk Alarms and Events, OPC UA ou OPC DA
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'CountOfTotal':2,
'CountOfInAlarm':0,
'CountOfUnacked':0,
'CountOfShelved':0,
'CountOfSuppressed':0,
'CountOfDisabled':0,
'CountOfBad':2
}
Génère :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir le nombre des 10 premières alarmes dont l'état est le plus fréquemment modifié au cours d'une période à partir du portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur IHM cible
begin_time : heure de début
end_time : heure de fin
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'Items':[{
'Proportion':0.8079424,
'Name':"WebPortalAlarm/Area1/FTAE1/Alarm1 LOLO",
'Quantity':33305
}]
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode permet d'obtenir le nombre et le taux des alarmes de chaque priorité sur une période donnée.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur cible FactoryTalk Alarms and Events, OPC UA ou OPC DA
begin_time : Heure de début
end_time : Heure de fin
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'Items':[{
'Priority':4,
'Quantity':18132,
'Proportion':0.996373236
}]
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir les 10 premières alarmes ayant la plus longue durée d'état active sur une période donnée.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur cible FactoryTalk Alarms and Events, OPC UA ou OPC DA
begin_time : Heure de début
end_time : Heure de fin
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Par exemple :
{
'Items':[{
'AlarmName':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Stale':164487,
}]
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour obtenir le nombre et la distribution des priorités des alarmes sur une période donnée.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
server_name : Nom du serveur cible FactoryTalk Alarms and Events, OPC UA ou OPC DA
begin_time : Heure de début, si le temps est inférieur à une heure, il sera calculé comme une heure
end_time : Heure de fin
headers : En-têtes de la demande
Renvoie :
Il renverra un ensemble de données au format JSON.
Chaque ensemble de données comprend plusieurs paramètres
Par exemple :
{
'Items':[{
'Time':"WebPortalAlarm/FTAE3/FTAE3_DigitalAlarm3",
'Urgent':164487,
'High':0,
'Medium':0,
'Low':0,
'UtcTime':"1678060800000"
}]
}
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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):
""" Cette méthode est utilisée pour déconnecter le portail d'état du système FactoryTalk.
Args:
Server_address : Demander l'adresse du serveur du portail d'état du système FactoryTalk
session_id : Renvoyer à partir des en-têtes de l'interface de connexion
Renvoie :
Génère :
Nous utilisons les protocoles réseau HTTP standard et le code de réponse est le code d'état standard renvoyé.
Par exemple :
Le respondCode 500 est une erreur réseau interne.
Le respondCode 401 n'est pas autorisé pour ce 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" # nom d'utilisateur de FactoryTalk
password = "xxx" # mot de passe de FactoryTalk
server_address = "https://" + server_ip + ":" + server_port + "/"# adresse du serveur Web état du système FT
# obtenir le jeton et l'ID de session dont nous avons besoin après la méthode de connexion
response_authentication_message = Login(server_address, username, password)
session_id = response_authentication_message[0]
token = response_authentication_message[1]
# Cette méthode est utilisée pour l'authentification lors du développement frontend
# Authentification(server_address,session_id)
# en-tête de demande avec jeton d'accès et ID de session
headers = GetRequestHeaders(token, session_id)
# obtenir la liste des applications en cours
response_application_list = Applications(server_address, headers)
target_hmi_application_name = "ClientTagApp1" # nom de l'application de démonstration à obtenir
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']
# obtenir la liste des serveurs en cours
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 est le nom du serveur de démonstration à obtenir
hmi_area = "/Area 1"
hmi_name = "/ClientTag1HMI1"
target_hmi_server = scope_ads_path + app_hmi_name + hmi_area + hmi_name
# serveur ftae cible sous la racine de l'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)
# # cette interface est utilisée pour obtenir les informations sur le serveur HMI actuellement connecté
Connections(server_address, hmi_server_name_encode, host_name_encode, headers)
# client IHM connecté au serveur IHM
client_name = 'DESKTOP-SS-1'
client_name_encode = Encode(client_name)
# cette interface est utilisée pour obtenir les informations historiques sur le serveur HMI actuellement connecté
ClientConnections(server_address, hmi_server_name_encode, host_name_encode, client_name_encode, headers)
ftae_server_name_encode = Encode(ftae_server_name)
# obtenir des informations de la base de données FTAE
DBName(server_address, ftae_server_name_encode, headers)
# obtenir le résumé des alarmes
AlarmSummary(server_address, ftae_server_name_encode, headers)
# définir le laps de temps comme la semaine dernière
begin_time = str(int(round(datetime.timestamp(datetime.now() + relativedelta(weeks=-2)) * 1000)))
# correspond à l'heure actuelle
end_time = str(int(round(time.time() * 1000)))
# utilisé pour interroger le nombre de changements d'état de l'alarme.
TopFrequency(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# utilisé pour obtenir le nombre de déclencheurs d'alarme pour chaque niveau dans le temps imparti
PriorityDistribution(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# utilisé pour obtenir les 10 premières alarmes les plus actives sur une période donnée
TopStaleness(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# utilisé pour obtenir le nombre d'alarmes déclenchées
Occurrences(server_address, ftae_server_name_encode, begin_time, end_time, headers)
# utilisé pour se déconnecter
LogOut(server_address, session_id)
if __name__ == '__main__':
main()
Fournir une réponse