接口与脚本：向第三方系统提供实时数据，并执行系统侧脚本任务

第三方使用 IO State API

进入“IO State API”，按页面显示的实际地址给第三方系统配置接口。IO State API 已经内置在 HQControl 主服务里，走网页同一个端口（默认 8080）。GET 适合按需查询，WebSocket 适合持续订阅实时变化。正式对接时要先确认访问 Key、订阅范围、完整点名、设备、功能组、设备组、信号分组和返回字段。

• 先在 Tag 管理里确认第三方要读的完整点名。
• 给 Key 授权范围时，功能组、设备组、信号组要写成“设备.组名称”（和报警一致，见下一条）；第三方查询时，设备、功能组、设备组、信号分组按字段分别填普通组名。
• 请求头使用 X-IO-State-Key，也可以按页面说明使用 accessKey 查询参数。
• 用 GET 读取少量点位，确认认证和返回字段。
• 再用 WebSocket 订阅实时变化，确认断线重连策略。
• 只把接口暴露给可信网络，不要直接裸露到公网。

# 以 IO State API 页面显示的实际地址为准
GET /api/v1/integration/io-state?fullCodes=DEVICE.RUN
GET /api/v1/integration/io-state?deviceCodes=PLC01
WebSocket /api/v1/integration/io-state/subscribe

给第三方分配 Key 的授权范围

在“IO State API”的「Key 管理」里给每个第三方分配独立 Key，并限定它能读哪些范围。授权按设备来：先选允许的设备，组（功能组、设备名称组）再绑定到具体设备，写成“设备编码.组编码”，和菜单、声音报警的范围写法一致，这样一个 Key 只会读到它被授权设备下的那些组，不会跨设备读到同名组。需要全开时，勾选“允许读取全部 IO State”。

• 客户端名称必填；授权范围至少填一项，或勾选“允许读取全部 IO State”。
• 允许设备：直接填设备编码，例如 DeviceA、DeviceB。
• 允许功能组 / 允许设备名称组：写成“设备编码.组编码”，把组绑定到具体设备（与菜单、声音报警一致），不要只写组名。
• 建议每个第三方一个独立 Key，并设置过期时间，方便单独停用或轮换。

接口筛选参数填写方式

IO State API 的筛选参数按字段维度填写，不按路径拼接。常用参数包括 fullCodes、deviceCodes、tagCodes、areaCodes、objectIdentifiers、groupCodes 和 qualities。多个值使用英文逗号分隔。对接前先用少量 fullCodes 确认返回结构，再扩大到设备、分组或质量范围。

• fullCodes 适合精确读取少量完整点名。
• deviceCodes 适合读取某台设备下的当前值。
• tagCodes 适合按 Tag 编码筛选，但仍要注意不同设备下可能存在同名 Tag。
• areaCodes、objectIdentifiers、groupCodes 适合按现场区域、对象和分组读取。
• qualities 适合筛选 good、bad、stale 等质量状态。
• 批量参数不要一次放入过大的范围；第三方系统应分页或分组读取。

GET /api/v1/integration/io-state?fullCodes=Line01.MainMotor.Running,Line01.MainMotor.SpeedPV
GET /api/v1/integration/io-state?deviceCodes=Line01&qualities=good
GET /api/v1/integration/io-state?groupCodes=Process&limit=100

脚本读写 Tag 用哪些方法

脚本里可以直接调用 HQControl 提供的 IO 方法，不需要额外导入模块。读取单点用 ReadTag 或类型化读取方法，批量读取用 ReadTags，写入用 WriteTag 或类型化写入方法。关键写入要写明原因，并确认目标 Tag 已允许控制。

• ReadTag(name, fallback=None) 返回完整点位对象，适合需要值、质量和时间戳的场景。
• ReadTags(filter=None, **kwargs) 用于批量读取，建议显式传入 limit。
• ReadTagBoolean、ReadTagInteger、ReadTagDouble、ReadTagString 用于按类型读取当前值。
• ReadTagQuality 和 ReadTagQualityGood 用于判断数据质量，不应替代报警规则。
• WriteTag、WriteTagBoolean、WriteTagInteger、WriteTagDouble、WriteTagString 用于写入系统 IO。

speed = ReadTagDouble("Line01.MainMotor.SpeedPV", 0.0)
quality = ReadTagQuality("Line01.MainMotor.SpeedPV", "bad")
items = ReadTags({"deviceCodes": ["Line01"], "limit": 50})
WriteTag("Line01.MainMotor.SpeedSP", 12.5, "number", reason="setpoint update")

常用 Python 脚本模板

这些模板放在宿主机 Integration Script 中，用于定时巡检、HTTP 通知、第三方系统同步和少量现场联动。正式上线前先手动运行一次，看输出、最近执行记录和失败日志；涉及写值时一定写明 reason，并确认目标 Tag 允许控制。

• 读取当前值：用 ReadTagBoolean / ReadTagDouble 等类型化方法，并给回退值。
• 质量保护：用 ReadTagQualityGood 判断数据是否可信，不要把 bad 质量当正常工艺值处理。
• HTTP 对接：安装包已内置 requests，调用外部系统时必须设置 timeout。
• 批量读取：ReadTags 要设置 deviceCodes、groupCodes 或 limit，避免一次读全站。

from datetime import datetime
import requests

tag_name = "Line01.MainMotor.SpeedPV"
speed = ReadTagDouble(tag_name, 0.0)
running = ReadTagBoolean("Line01.MainMotor.Running", False)
quality_good = ReadTagQualityGood(tag_name)
now_text = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

if not quality_good:
    print(f"{now_text} {tag_name} quality is bad, skip action")
elif running and speed > 80:
    payload = {
        "tag": tag_name,
        "speed": speed,
        "time": now_text,
        "message": "main motor speed high",
    }
    requests.post("http://127.0.0.1:9000/api/alarm", json=payload, timeout=3)

# 批量读一组点位，适合巡检或报表前置数据
items = ReadTags({"deviceCodes": ["Line01"], "limit": 100})
print(f"read {len(items or [])} tags at {now_text}")

# 需要写值时保留原因，便于审计
WriteTag("Line01.MainMotor.SpeedSP", 12.5, "number", reason="integration script setpoint update")

Python 获取网络内容与读写 SQL 示例

宿主机脚本常见任务是读取外部 HTTP/JSON、把数据同步进业务库，或从外部 SQL 表取配方、计划、班次等信息。现场脚本要控制超时、使用参数化 SQL、关闭连接，不要把数据库密码硬编码到脚本里；建议从环境变量、脚本连接配置或受控配置文件读取。

• 获取网络内容：用 requests.get(..., timeout=3)，并调用 raise_for_status() 让 HTTP 错误尽早暴露。
• 读 SQL：用参数化查询，不拼接用户输入；查询结果为空时给出清晰输出。
• 写 SQL：用事务提交，写入前准备好 Tag 值、质量和时间戳。
• 数据库驱动：PostgreSQL 可用 psycopg，MySQL 可用 pymysql，ODBC/SQL Server 可用 pyodbc。

import os
from datetime import datetime
import requests
import psycopg

# 1) 获取网络 JSON 内容
resp = requests.get('https://example.com/api/shift-plan', timeout=3)
resp.raise_for_status()
plan = resp.json()
target_speed = float(plan.get('targetSpeed', 0))

# 2) 读取 HQControl 当前 Tag
tag_name = 'Line01.MainMotor.SpeedPV'
speed = ReadTagDouble(tag_name, 0.0)
quality = ReadTagQuality(tag_name, 'bad')
sample_time = datetime.now()

# 3) 读写 PostgreSQL，连接串建议来自环境变量或脚本连接配置
dsn = os.getenv('MES_PG_DSN', 'host=127.0.0.1 port=5432 dbname=mes user=mes password=change-me')
with psycopg.connect(dsn) as conn:
    with conn.cursor() as cur:
        cur.execute(
            'select order_no, product_code from work_orders where line_code = %s order by created_at desc limit 1',
            ('Line01',),
        )
        row = cur.fetchone()
        order_no = row[0] if row else ''

        cur.execute(
            'insert into hq_samples(tag_name, value, quality, target_speed, order_no, sampled_at) values (%s, %s, %s, %s, %s, %s)',
            (tag_name, speed, quality, target_speed, order_no, sample_time),
        )
    conn.commit()

print(f'synced {tag_name}={speed}, quality={quality}, target={target_speed}')