所属产品: Mingcedar CONTROL

IO State API：授权、精确读取与实时订阅 | Mingcedar

准备启用 IO State 的 HTTPS 地址、质量正常的点位及具备 io:manage 的账号。外部客户端仅读取已采集状态；API 不连接 PLC，也不提供写入、报警或历史趋势。

首次发布 2026-10-25T12:48:42.552057056Z 最近更新 2026-10-25T12:48:42.552056305Z

<div class="manual-doc-article manual-doc-article-section"><h2 id="io-state-api">开发教程: IO State API：授权、精确读取与实时订阅</h2>

创建 Profile 和 Access Key，读取 SignalRef，获取单次 Ticket 并订阅 WebSocket。

<h3 id="io-state-api-item-1">准备条件与接口职责</h3>

准备启用 IO State 的 HTTPS 地址、质量正常的点位及具备 io:manage 的账号。外部客户端仅读取已采集状态；API 不连接 PLC，也不提供写入、报警或历史趋势。

<ul>

<li>SignalRef 使用目录中的 hq://io/...，派生位使用 hq://io-bit/...；不要把设备显示名当作稳定标识。</li>

<li>计算 Tag 不属于本教程 IO State V2 SignalRef 范围，应使用计算 Tag 专用消费接口。</li>

<li>Access Key 是服务器端客户端凭据，不应嵌入公开 HTML、日志、画面记录或分享的代码。自写运行页面使用页面 SDK。</li>

</ul>

</article>

<h3 id="io-state-api-item-2">1. 创建只包含一个信号的 Profile</h3>

Profile 定义交付信号。首个连接使用一个质量为 good 的物理 Tag 完整 SignalRef。

<ol>

<li>进入 IO State API → Subscription Profiles，点击“新建 Profile”。</li>

<li>名称填 Equipment Status，代码填 equipment_status；成员模式选 frozen，勾选启用 Profile。</li>

<li>点击“添加 Selector”，类型选择 tagRef。勾选“包含”及“必须匹配”，在 SignalRef 粘贴完整 hq://io/... URI。</li>

<li>首个单点练习保留高级 delivery、projection、limits 的默认值。点击“保存草稿”。</li>

<li>在 Profile 列表选中刚保存的行，打开详情并点击“预览（Preview）”。</li>

</ol>

完成标志 预览应解析出 1 个信号且允许发布；草稿不可订阅。

<div class="doc-step-troubleshooting">如果结果不一致<ul>

<li>解析为 0 或必须匹配失败：检查 SignalRef 是否完整且属于当前可用目录，不用设备显示名称替换 UUID。</li>

<li>不能发布：阅读预览提示并修正 Selector，再保存并重新预览。</li>

</ul></div>

</article>

<h3 id="io-profile-publish">2. 发布 Profile 并确认 Revision</h3>

发布使用已保存、已预览的草稿。Revision 标识客户端应该订阅的确定版本。

<ol>

<li>在通过的预览结果下点击“发布 Revision（Publish Revision）”。</li>

<li>核对确认窗口中的信号数量为 1，点击确认发布。</li>

<li>回到详情，检查状态为 Published 且 Current Revision 有值；打开 Revision 历史核对刚发布的记录。</li>

<li>记录 Profile 的代码和 Revision。后续客户端查询会返回 Profile ID，代码与 ID 不可混用。</li>

</ol>

完成标志 已生成 Revision。若草稿被他人修改，刷新并重新预览，不能覆盖版本冲突。

</article>

<h3 id="io-state-api-item-3">3. 为外部客户端生成 Access Key</h3>

Access Key 标识调用方。此步骤创建凭据；下一步还需设置 V2 授权范围。

<ul>

<li>请求头为 X-IO-State-Key，不是 Portal 管理员登录 Token。</li>

<li>可用 Profile 返回 id、revision、resolvedCount、membershipHash、streamAllowed 和 adHocReadAllowed。检查模式开关后再调用。</li>

<li>revision、计数与序列等指定字段采用十进制字符串；不要转换为可能失真的 JavaScript Number。</li>

</ul>

<ol>

<li>进入 Access Keys，点击“新建 Access Key”。客户端名称填 Equipment Reporting。</li>

<li>按项目要求填写过期时间，勾选“立即启用”和“允许临时查询”，并发订阅上限先选 1。</li>

<li>选择“指定范围”，类型选 Tags，填写同一目标的完整设备编码.Tag 编码并点击“添加范围”。创建表单至少需要一项范围；此处不填 hq:// URI，也不选择“全部 IO State”。</li>

<li>确认列表只包含本次目标点。创建后还需按下一步绑定 Profile 并添加对应 UUID 的 V2 原生 Scope；表单中的点分编码不能替代 V2 授权。</li>

<li>点击“生成 Key”，在“完整 Key 仅本次可见”处复制并保存到客户端凭据管理位置。</li>

</ol>

完成标志 客户端已启用，完整 Key 已保存；摘要不能用于请求。

</article>

<h3 id="io-native-scope">4. 绑定 Profile 并添加原生 TagRef Scope</h3>

未启用“全部 IO State”时，订阅信号须同时位于已绑定 Profile 和原生 Scope 内；仅绑定 Profile 不授予范围。

<ol>

<li>在台账选中 Equipment Reporting，找到“V2 授权”，点击“编辑 V2 授权”。</li>

<li>在 Subscription Profiles 勾选 Equipment Status。</li>

<li>在 Native TagRef Scopes 中选择 Tag 类型，分别填写该 SignalRef 对应的 IO Domain UUID、Device UUID 和 Tag UUID。使用目录的真实身份，不填 Code。</li>

<li>点击“添加（Add）”，确认该条 Scope 已进入下方列表，再点击“保存 V2 授权”。</li>

<li>关闭编辑器后核对详情：绑定 Profile 数量为 1，原生 Scope 数量为 1，身份与目标点位一致。</li>

</ol>

完成标志 Key 同时具有 Profile 绑定和单点 Scope。修改授权会终止此 Key 的现有流，客户端需要重新连接。

<div class="doc-step-troubleshooting">如果结果不一致<ul>

<li>详情显示 0 个原生 Scope：仅填写输入框还未添加；返回编辑器点击 Add 后保存。</li>

<li>列表有 Profile 但读取被拒绝：核对 Scope 的三个身份和“允许临时查询”，不要直接扩大到全量权限。</li>

</ul></div>

</article>

<h3 id="io-client-setup">5. 建立外部客户端目录</h3>

文件放在调用方电脑或服务容器，不放入 Integration Script。启动器提示输入地址和 Key，Key 不回显。

<ol>

<li>在外部客户端建立 io-client 文件夹，把下方代码保存为 run_io_client.py。</li>

<li>将下一节读取代码保存为同一目录下的 io_read.py。两个文件名必须一致，不使用 .txt 扩展名。</li>

<li>打开该目录的终端，执行 python run_io_client.py。Windows 若使用 py 启动器，可执行 py -3 run_io_client.py。</li>

<li>输入实际 HTTPS 根地址，再输入完整 Access Key。程序先显示此 Key 可用的 Profile、ID、revision 和权限开关。</li>

<li>出现 Mode 提示时输入 read，再粘贴与 Scope 对应的完整 SignalRef。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">run_io_client.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: run_io_client.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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(

)

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()

</code></pre></div>

完成标志 Profile 列表出现 Equipment Status 对应记录；读取练习需要 adHocReadAllowed，订阅练习需要 streamAllowed。完整 Key 不出现在输出中。

</article>

<h3 id="io-state-api-item-4">6. 保存读取程序并核对响应</h3>

将 io_read.py 与 run_io_client.py 放在同目录，用启动器 read 模式运行。启动器设置变量并调用 main()；程序仅用标准库。

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">io_read.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: io_read.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># External client: io_read.py (standard library only)

import json

import os

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:

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()

</code></pre></div>

完成标志 输出包含目标 signalRef、值类型 kind、value、quality=good 和 observedAt。与 Tag 测试比较同一时段的数据。

</article>

<h3 id="io-state-api-item-5">请求和响应字段</h3>

POST /api/v2/integration/io-state/read 使用 JSON 对象，signalRefs 为 1–1000 项，include 可选 ["metadata"]。响应 type 为 io_state_read，items 为逐输入顺序的结果；单项失败不等于整批 HTTP 失败。

<ul>

<li>每项包含 signalRef、quality、qualityDetail；有状态时提供 state，无数据时可能省略 state。</li>

<li>state.value 是 {kind, value}，不是裸数值。int64/uint64/decimal 的字符串表示应保持精度。</li>

<li>state 包括质量、观测时间、服务端时间、版本与来源。不要把 missing、stale、pending、unavailable 转成数值 0。</li>

<li>预期结果：控制台输出当前 SignalRef、值类型、值、good 质量和观测时间，与工作台诊断一致。</li>

</ul>

</article>

<h3 id="io-state-api-item-6">7. 安装客户端依赖并观察实时流</h3>

先完成读取练习。此程序仅观察连接帧，不维护生产缓存；Access Key 无需写入代码。

<ol>

<li>在运行客户端的同一个 Python 环境执行 python -m pip install "websocket-client==<approved-version>"，先替换版本占位符。离线客户端使用前文 Wheel 准备方法安装。</li>

<li>把下方代码保存为 io_stream.py，放到两个客户端文件旁边。</li>

<li>再次执行 python run_io_client.py，输入地址和 Key，在 Profile 列表记录目标 id 与 revision。</li>

<li>Mode 输入 stream，Published Profile ID 输入刚返回的 id，而不是 equipment_status 代码。</li>

<li>观察 subscription_ready、snapshot_begin、snapshot_chunk、snapshot_end，随后观察 heartbeat 或 io_state_delta。按 Ctrl+C 结束。</li>

</ol>

<div class="manual-code" data-code-language="python"><div class="manual-code-toolbar">io_stream.pyPython<button type="button" data-copy-code data-copied="已复制" data-failed="请选中代码后复制" aria-label="复制代码: io_stream.py">复制代码</button></div><pre tabindex="0" class="docs-code-block shiki github-light" style="background-color:#fff;color:#24292e" tabindex="0"><code># 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

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):

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()

main()

</code></pre></div>

完成标志 应收到完整快照及持续心跳；值未变化时，没有 delta 属于正常情况。

<div class="doc-step-troubleshooting">如果结果不一致<ul>

<li>StopIteration：客户端可见 Profile 中找不到输入的 ID，检查绑定、已发布状态及输入值。</li>

<li>ModuleNotFoundError: websocket：使用运行启动器的同一解释器执行 pip 安装，发行包名为 websocket-client。</li>

<li>Ticket 失败或连接断开：退出本次诊断，核对 Revision、授权及错误码后重新启动，申请新 Ticket。</li>

</ul></div>

</article>

<h3 id="io-state-api-item-7">实时订阅的数据处理规则</h3>

帧顺序：subscription_ready → snapshot_begin → snapshot_chunk（可多次）→ snapshot_end → io_state_delta / heartbeat。示例仅显示帧；生产客户端须按下列规则维护缓存。

<ul>

<li>Ticket 请求必须包含 profileId、expectedRevision、clientCapabilities；后者包含 maxSignals、maxFrameBytes、maxUpdatesPerSecond。容量数值应与实际 Profile 匹配。</li>

<li>Ticket 为单次使用，受过期时间、IP 与 User-Agent 约束；不要记录含 Ticket 的 URL，连接失败后重新申请。</li>

<li>生产客户端以 snapshotId 累积临时快照，检查 chunkIndex 顺序、chunks、items、complete、Profile revision 和 membershipHash，完整后才替换活动缓存。</li>

<li>每帧检查 streamId、streamEpoch、frameSeq 连续性；版本中的 domainEpoch/changeSeq 用于区分信号状态代次。</li>

<li>io_state_delta 按 ops 处理 upsert/remove；heartbeat 更新连接可用性，不是新采集值；不要把最新状态流当成每次变化都不丢的事件日志。</li>

<li>收到 reset_required、error、序列缺口、成员变更或连接中断时，将旧缓存标为不可用，关闭旧连接，核对授权与版本，再退避申请新 Ticket 和完整快照。</li>

</ul>

</article>

<h3 id="io-state-api-item-8">连接排查与重连处理</h3>

按授权、范围、Profile 版本、客户端容量、连接及数据质量的顺序核查。

<ul>

<li>401：Key 缺失、失效，或 Ticket 无效/过期/已使用；不要把凭据打印到错误日志。</li>

<li>403：读取或订阅范围不足；409：Profile revision 已变化，重新读取当前发布状态。</li>

<li>413：缩小单次读取或 Profile/帧容量；429：遵循限流提示；503/504：检查服务与源数据可用性。</li>

<li>TLS 错误：安装正确 CA 信任链，使用证书匹配的主机名，不关闭验证。</li>

<li>无数据、源失联、Key 撤销或 Profile 变更时，客户端应显示数据不可用并处理重连，不继续把旧值显示为实时 good。</li>

</ul>

</article>

</div>

图示 1: 配置订阅 Profile

Profile 中设置名称、代码和成员模式，再用 Selector 限定需要共享的信号。发布前需预览解析结果。

</figcaption>

</figure>

</div>