准备启用 IO State 的 HTTPS 地址、质量正常的点位及具备 io:manage 的账号。外部客户端仅读取已采集状态；API 不连接 PLC，也不提供写入、报警或历史趋势。
准备启用 IO State 的 HTTPS 地址、质量正常的点位及具备 io:manage 的账号。外部客户端仅读取已采集状态；API 不连接 PLC，也不提供写入、报警或历史趋势。
Profile 定义交付信号。首个连接使用一个质量为 good 的物理 Tag 完整 SignalRef。
完成标志 预览应解析出 1 个信号且允许发布；草稿不可订阅。
如果结果不一致
发布使用已保存、已预览的草稿。Revision 标识客户端应该订阅的确定版本。
完成标志 已生成 Revision。若草稿被他人修改，刷新并重新预览，不能覆盖版本冲突。
Access Key 标识调用方。此步骤创建凭据；下一步还需设置 V2 授权范围。
完成标志 客户端已启用，完整 Key 已保存；摘要不能用于请求。
未启用“全部 IO State”时，订阅信号须同时位于已绑定 Profile 和原生 Scope 内；仅绑定 Profile 不授予范围。
完成标志 Key 同时具有 Profile 绑定和单点 Scope。修改授权会终止此 Key 的现有流，客户端需要重新连接。
如果结果不一致
文件放在调用方电脑或服务容器，不放入 Integration Script。启动器提示输入地址和 Key，Key 不回显。
# External client: run_io_client.py
import getpass
import json
import os
from urllib.request import Request, urlopen
os.environ["HQCONTROL_URL"] = input("Mingcedar HTTPS URL: ").strip().rstrip("/")
if not os.environ["HQCONTROL_URL"].startswith("https://"):
raise ValueError("Enter the configured HTTPS URL")
os.environ["HQCONTROL_IO_KEY"] = getpass.getpass("Access Key: ")
request = Request(
os.environ["HQCONTROL_URL"] + "/api/v2/integration/io-state/profiles",
headers={"X-IO-State-Key": os.environ["HQCONTROL_IO_KEY"]},
)
with urlopen(request, timeout=10) as response:
print(json.dumps(json.load(response), indent=2))
mode = input("Mode (read/stream): ").strip()
if mode == "read":
os.environ["HQCONTROL_SIGNAL_REF"] = input("SignalRef: ").strip()
from io_read import main
elif mode == "stream":
os.environ["HQCONTROL_PROFILE_ID"] = input("Published Profile ID: ").strip()
from io_stream import main
else:
raise ValueError("Choose read or stream")
main()
完成标志 Profile 列表出现 Equipment Status 对应记录；读取练习需要 adHocReadAllowed，订阅练习需要 streamAllowed。完整 Key 不出现在输出中。
将 io_read.py 与 run_io_client.py 放在同目录，用启动器 read 模式运行。启动器设置变量并调用 main()；程序仅用标准库。
# External client: io_read.py (standard library only)
import json
import os
from urllib.request import Request, urlopen
BASE_URL = os.environ["HQCONTROL_URL"].rstrip("/")
ACCESS_KEY = os.environ["HQCONTROL_IO_KEY"]
SIGNAL_REF = os.environ["HQCONTROL_SIGNAL_REF"]
def main():
request = Request(
BASE_URL + "/api/v2/integration/io-state/read",
data=json.dumps({"signalRefs": [SIGNAL_REF], "include": ["metadata"]}).encode(),
headers={"X-IO-State-Key": ACCESS_KEY, "Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=10) as response:
result = json.load(response)
if result.get("type") != "io_state_read" or len(result.get("items", [])) != 1:
raise RuntimeError("Unexpected read response")
item = result["items"][0]
if item.get("quality") != "good" or "state" not in item:
raise RuntimeError("Signal unavailable: " + item.get("qualityDetail", "unknown"))
state = item["state"]
typed = state["value"]
# Preserve int64 / uint64 / decimal strings; do not convert them to float.
print(json.dumps({"signalRef": item["signalRef"], "kind": typed["kind"],
"value": typed["value"], "quality": state["quality"],
"observedAt": state["observedAt"]}, ensure_ascii=False))
if __name__ == "__main__":
main()
完成标志 输出包含目标 signalRef、值类型 kind、value、quality=good 和 observedAt。与 Tag 测试比较同一时段的数据。
POST /api/v2/integration/io-state/read 使用 JSON 对象，signalRefs 为 1–1000 项，include 可选 ["metadata"]。响应 type 为 io_state_read，items 为逐输入顺序的结果；单项失败不等于整批 HTTP 失败。
先完成读取练习。此程序仅观察连接帧，不维护生产缓存；Access Key 无需写入代码。
# External client: io_stream.py. Install the approved websocket-client package.
# This diagnostic prints the stream lifecycle; it is not a production state cache.
import json
import os
from urllib.parse import urlencode
from urllib.request import Request, urlopen
import websocket
BASE = os.environ["HQCONTROL_URL"].rstrip("/")
KEY = os.environ["HQCONTROL_IO_KEY"]
PROFILE = os.environ["HQCONTROL_PROFILE_ID"]
USER_AGENT = "Mingcedar-Tutorial/1.0"
def request_json(path, body=None):
headers = {"X-IO-State-Key": KEY, "User-Agent": USER_AGENT}
data = None
if body is not None:
headers["Content-Type"] = "application/json"
data = json.dumps(body).encode()
with urlopen(Request(BASE + path, data=data, headers=headers), timeout=10) as response:
return json.load(response)
def main():
if not BASE.startswith("https://"):
raise ValueError("Use the configured HTTPS server URL")
profiles = request_json("/api/v2/integration/io-state/profiles")["items"]
profile = next(item for item in profiles if item["id"] == PROFILE)
ticket = request_json("/api/v2/integration/io-state/stream-tickets", {
"profileId": profile["id"], "expectedRevision": profile["revision"],
"clientCapabilities": {"maxSignals": 1000, "maxFrameBytes": 1048576,
"maxUpdatesPerSecond": 1000},
})
address = "wss://" + BASE[len("https://"):] + "/api/v2/integration/io-state/subscribe?" + urlencode({"ticket": ticket["ticket"]})
connection = websocket.create_connection(address, timeout=30, suppress_origin=True,
header=["User-Agent: " + USER_AGENT])
stream = None
previous = None
try:
while True:
frame = json.loads(connection.recv())
identity = (frame["streamId"], frame["streamEpoch"])
sequence = int(frame["frameSeq"])
if stream is not None and (identity != stream or sequence != previous + 1):
raise RuntimeError("Stream continuity lost; obtain a new ticket and snapshot")
stream, previous = identity, sequence
print(frame["type"], frame["frameSeq"])
if frame["type"] in ("reset_required", "error"):
raise RuntimeError("Stream reset; review profile and authorization before reconnecting")
finally:
connection.close()
if __name__ == "__main__":
main()
完成标志 应收到完整快照及持续心跳；值未变化时，没有 delta 属于正常情况。
如果结果不一致
帧顺序：subscription_ready → snapshot_begin → snapshot_chunk（可多次）→ snapshot_end → io_state_delta / heartbeat。示例仅显示帧；生产客户端须按下列规则维护缓存。
按授权、范围、Profile 版本、客户端容量、连接及数据质量的顺序核查。
Profile 中设置名称、代码和成员模式，再用 Selector 限定需要共享的信号。发布前需预览解析结果。