当前位置：首页 > news >正文

西门子WinCC Unified PC的GraphQL使用手册

news 来源：原创 2025/5/15 6:32:46

TIA V20版本：添加用户
在这里插入图片描述

添加角色，并充分授权，尤其是GraphQL的读写权限。
在这里插入图片描述
通过SIMATIC Runtime Manager启动wincc unifi工程。

打开浏览器，访问本地的https://localhost/graphql/，运行正常如图：

连接外网，打开https://studio.apollographql.com/sandbox/explorer工具，在线调试GraphQL接口：
在这里插入图片描述
打开查询指令窗口

第一步，请求login获取token:

mutation{
login(username:"wincc",password:"Wincc12345"){
token
user{
fullName
id
}
error{
code
description
}
}
}

在这里插入图片描述
获取token，后面需要添加到headers里：Authorization Bearer e68814f546ed1360cb533ed7ecf77ae0。
如图所示：

第二步：查询获取变量值Tag_2是变量名称。

query Query{
tagValues(names:["Tag_2"]){
name
value{
value
timestamp
quality{
quality
subStatus
}
}
error{
code
description
}
}
}

在这里插入图片描述
第三步，写入一个值0：

mutation exampleTagValueWrite {
writeTagValues(input:[
{
name:"Tag_2",
value: "0"
}
], quality:{quality: GOOD_NON_CASCADE}){
name
error{
code
description
}
}
}

在这里插入图片描述
第四步，订阅一个变量：

subscription subscription{
tagValues(names:["Tag_2"]){
value{
value
timestamp
quality{
quality
limit
subStatus
}
}
error{
description
}
}
}
}

在这里插入图片描述
GraphQL通过python客户端读取和写入变量:

from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransporturl = 'http://localhost:4000/graphql'
USERNAME = "wincc"
PASSWORD = "Wincc12345"
TAG_NAME = "Tag_2" 
transport = RequestsHTTPTransport(url=url,verify=False, retries=3)
client = Client(transport=transport, fetch_schema_from_transport=True)query = '''
mutation{
login(username:"%s",password:"%s"){
token
user{
fullName
id
}
error{
code
description
}
}
}
'''variable_values = {'Authorization': 'Bearer 2d263aa90155e66bb24f7a4604153ee7'}  # 如果有变量的话
result = client.execute(gql(query % (USERNAME, PASSWORD)), variable_values=variable_values)
token = result['login']['token']
headers={'Authorization': 'Bearer ' + token}
transport = RequestsHTTPTransport(url=url,headers=headers,verify=False, retries=3)
client = Client(transport=transport, fetch_schema_from_transport=True)query = '''
mutation exampleTagValueWrite {
writeTagValues(input:[
{
name:"%s",
value: "0"
}
], quality:{quality: GOOD_NON_CASCADE}){
name
error{
code
description
}
}
}
'''
result = client.execute(gql(query))
print(result)
query = '''
query Query{
tagValues(names:["%s"]){
name
value{
value
timestamp
quality{
quality
subStatus
}
}
error{
code
description
}
}
}
'''
result = client.execute(gql(query % (TAG_NAME)))
print(result)

GraphQL通过python客户端订阅变量:

import asyncio
from gql import Client, gql
from gql.transport.websockets import WebsocketsTransport
from websockets.exceptions import ConnectionClosedGRAPHQL_WS_URL = "wss://localhost:4000/graphql"
TAG_NAME = "Tag_2" 
AUTH_TOKEN = "1a64a7289995f7da6a9baa0cd5eb93db"headers = {'Authorization': 'Bearer ' + AUTH_TOKEN}subscription = gql("""subscription {tagValues(names: ["%s"]) {namevalue {value}notificationReason}}
""" % TAG_NAME)async def main():transport = WebsocketsTransport(GRAPHQL_WS_URL, init_payload=headers)counter = 0try:# Using `async with` on the client will start a connection on the transport# and provide a `session` variable to execute queries on this connectionasync with Client(transport=transport,fetch_schema_from_transport=True,) as session:async for result in session.subscribe(subscription):print(result)counter += 1if counter == 1000:print ("Closing connection from the client")          breakexcept ConnectionClosed:print ("Connection closed by the server")asyncio.run(main())