跳至主要内容

Python 项目与模块

先确认 Script 服务可用。代码管理需超级管理员及 integration:manage;Run Test 另需 integration:execute;Python 模块管理另需 system:view 和 system:manage。按顺序完成计算、依赖安装、事件调用与 Tag 读写练习。

开发教程

开发教程: Python 项目与模块

使用 CPython 3.13 创建项目、复用函数、安装依赖、读取 Tag、配置事件及受控写入。

开始前准备

先确认 Script 服务可用。代码管理需超级管理员及 integration:manage;Run Test 另需 integration:execute;Python 模块管理另需 system:view 和 system:manage。按顺序完成计算、依赖安装、事件调用与 Tag 读写练习。

  • 代码(Code):创建项目、编辑文件;工具栏的“Run Test”执行当前选中的 Python 文件,未保存时显示“Save & Test”。
  • 运行(Run):配置何时触发的事件,以及事件执行的动作。记录(Records):检查正式事件执行结果。
  • Settings → Python → Python 模块:选择项目并管理模块;高级项目配置管理入口、运行限制、服务器离线目录、环境变量和 Secret。
  • 本教程中的尖括号是待替换项;文件名、函数名和练习用数值则按示例填写。

1. 创建项目和两个 Python 文件

位置:配置中心 → script → 代码(Code)。先创建项目,再添加文件。

  1. 点击“管理项目(Manage Projects)”,在弹窗内点击“新建(New)”。
  2. 在“项目名称”填写 Equipment Integration,点击“创建项目(Create Project)”。关闭弹窗,在代码页项目下拉框选择刚创建的项目。
  3. 在 common/ 行点击新建文件的“+”,输入 scaling,点击“创建”。系统补齐 .py。
  4. 在 scripts/ 行以相同方式创建 main。若 main.py 已存在,直接选中它,不重复创建。
  5. 检查编辑器标题中的完整路径,确认分别是 common/scaling.py 与 scripts/main.py。
TextText
Project/
├── scripts/
│   ├── __init__.py
│   └── main.py
└── common/
    ├── __init__.py
    └── scaling.py

完成标志 两个文件均可选中,路径正确。下方目录仅说明文件结构。

如果结果不一致

  • 没有创建按钮:核对当前账号的脚本管理权限。文件出现在错误目录时,先修正位置再复制代码。

2. 保存公共函数

将输入量程 0–100 换算为 0–10。

  1. 选中 common/scaling.py,确认编辑器路径。
  2. 用下方完整代码替换文件内容,点击“保存(Save)”。
  3. 等待未保存标记消失,再切换文件。本文件只定义函数,不需要单独运行。
common/scaling.pyPython
# common/scaling.py
def scale_linear(value, input_min, input_max, output_min, output_max):
    if input_max == input_min:
        raise ValueError("Input range must not be zero")
    return output_min + (value - input_min) * (output_max - output_min) / (input_max - input_min)

完成标志 保存后无输出;下一步由 main.py 调用。

3. 调用公共函数并核对控制台

事件调用 main(ctx);代码页通过 __main__ 分支调用同一函数。

  1. 选中 scripts/main.py,粘贴下方完整代码并保存。
  2. 保持选中 main.py,点击工具栏 Run Test。不要运行 scaling.py。
  3. 在编辑器下方“控制台(Console)”查看执行状态和输出。
  4. 将 main({"raw": 50}) 中的 50 改为 25,保存并再次运行,核对换算值变成 2.5;完成后改回 50。
scripts/main.pyPython
# scripts/main.py
from ..common.scaling import scale_linear

def main(ctx):
    value = scale_linear(float(ctx.get("raw", 50)), 0, 100, 0, 10)
    print("Scaled value:", value)
    return {"scaled": value}

if __name__ == "__main__":
    main({"raw": 50})

完成标志 首次输出 Scaled value: 5.0,状态为 Completed;修改输入后输出 2.5。

如果结果不一致

  • Completed 但没有输出:检查选中文件,以及底部是否包含 if __name__ == "__main__"。
  • ImportError:确认两个文件属于同一项目,并保持 scripts/ 和 common/ 的层级;不要从操作系统终端直接运行本示例。

公共函数与自写包的导入规则

scripts/main.py 使用 from ..common.scaling import scale_linear。若文件位于 scripts/controls/main.py,应使用三个点返回项目包根。绝对导入必须使用项目清单中的实际 package 名称。

  • 同一项目复用:common/ 随项目一起构建和应用,使用相对导入,无需在 Python 模块页面安装。修改公共函数后保存,并确认运行版本已应用。
  • 跨项目复用:将自写代码制作为有明确版本和依赖声明的 Wheel,按下文流程分别安装到需要它的项目。安装后按包内实际模块名普通 import,不使用 Wheel 文件名。
  • 不要修改 sys.path 指向宿主目录,也不要将另一个项目的脚本路径作为导入来源。

完成标志 修改并保存 scaling.py 后,重新运行 main.py 验证结果。跨项目复用须安装依赖,不能直接引用另一项目路径。

4. 记录本项目的 Python 运行目标

依赖按项目安装,不共享准备机或服务器系统 Python 的包。Python 模块页需上述系统设置权限。

  • 标准库 json、csv、sqlite3、datetime、urllib、smtplib 无需安装。
  • 常用内置模块包括 requests、psycopg、PyMySQL、pyodbc、paho-mqtt、websocket-client 和 cryptography。新项目首次构建环境时安装内置模块;已有项目可通过“添加内置模块”补充,已选版本会保留。
  • Excel 等额外用途可按需安装 openpyxl 等模块。使用 pyodbc 连接数据库时,服务器还需安装对应的 ODBC 驱动。
  • 发行包名和 import 名可能不同:PyMySQL → pymysql,websocket-client → websocket。
  1. 打开 Settings → Python → Python 模块,选择 Equipment Integration。
  2. 查看模块列表上方的“运行环境”,记录 Python 版本、操作系统和架构;当前目标为 Python 3.13。
  3. 查看模块列表中的名称和版本,确认项目是否已有本练习所需模块;列表标题会随项目状态显示为“项目模块”“已准备模块”或“已安装模块”。
  4. Linux 容器以容器内运行目标为准;Windows、Linux 原生部署使用各自运行目标。

完成标志 按已确认的项目运行目标选择内置模块或兼容 Wheel。

5. 选择内置模块或准备 Wheel 及其依赖

内置模块无需下载。其他依赖在匹配目标 OS、CPU 架构和 Python 3.13 的准备机上,按批准的精确版本下载。下列命令仅在准备机执行。

  1. 先查看项目模块列表。若 psycopg 和 PyMySQL 已提供为内置模块,可点击“添加内置模块”,核对预览中全部拟添加的包,再跳到第 7 步;无需重复下载。
  2. 需要其他第三方库或指定版本时,在与服务器系统、架构和 Python 版本一致的可联网准备机执行 python --version。本例为后续 SQL 练习准备 psycopg 和 PyMySQL;只使用一种数据库时,可移除另一种包及其导入检查。
  3. 在下方命令中分别替换两处 <approved-version> 为选定包的精确版本。引号保留,不要把尖括号原样运行。
  4. 先执行 pip download,等待命令成功,再检查 ./wheelhouse 中包含主包及依赖包的 .whl 文件。
  5. 自写包也应构建为带明确版本和依赖声明的 .whl;可在临时虚拟环境执行下方离线安装与 import 检查,确认依赖完整和实际导入名后再上传。
ShellShell
# Run on a preparation machine matching the target OS and CPU, with Python 3.13.
# Replace the version placeholders with your approved exact versions.
python -m pip download --only-binary=:all: --dest ./wheelhouse "psycopg[binary]==<approved-version>" "PyMySQL==<approved-version>"
python -m pip hash "<wheel-file>"

# Optional local verification in a disposable preparation environment:
python -m venv ./dependency-check
# Use dependency-check/bin/python (Linux) or dependency-check\Scripts\python.exe (Windows):
<venv-python> -m pip install --no-index --find-links=./wheelhouse "psycopg[binary]==<approved-version>" "PyMySQL==<approved-version>"
<venv-python> -c "import psycopg, pymysql; print('Dependencies ready')"

完成标志 内置模块已列入预览,或目录已有所需 .whl。离线检查应输出 Dependencies ready;仍须在目标项目验证。

6. 上传 Wheel 并检查安装预览

优先使用与目标一致的准备环境。必须交叉下载时,明确指定 --platform、--python-version 3.13、--implementation cp、--abi cp313 和 --only-binary=:all:;平台标签必须匹配目标运行环境,不能仅凭 Linux 名称推定 glibc/musl 兼容性。

  1. 核对每个 Wheel 的发行包、版本、Python/ABI/平台标签及 SHA-256;同一发行包只保留目标版本,并包含全部间接依赖。
  2. 在 Settings → Python → Python 模块选择当前项目,点击“添加”,选中所需 .whl 文件;也可上传仅包含完整 Wheel 依赖集合的 ZIP,或把文件拖入模块列表。
  3. 查看“确认安装内容”预览,核对新增或更新的包及版本;若提示缺少依赖,补齐 Wheel 后继续添加,暂不提交安装。
  4. 仅在需要从服务器目录导入时,展开“高级项目配置 → 服务器离线目录”,把 .whl 普通文件复制到显示的项目目录并点击“刷新清单”;再回到模块列表展开“离线目录导入”,点击“检查目录中的包”。该目录不接收 ZIP、源码包、子目录或符号链接。
  5. Linux 容器采用容器内路径,并确保服务账号可读取;刷新清单仅检查目录,仍须完成安装预览和下一步的安装。

完成标志 预览列出预期变更,无兼容性、版本或缺失依赖错误。项目运行环境尚未改变。

7. 安装模块并实际导入

核对预览后提交。新项目可先安装模块供代码页运行;已有运行版本时,检查通过后应用新版本。

  1. 核对预览后,若所需模块已在当前项目的模块列表中,直接到代码页做导入检查;否则按页面显示点击“安装到项目”或“安装并应用”。前者准备尚无运行版本的项目供手动运行,后者检查并应用新的运行版本。
  2. 若提交安装,等待“模块已安装到项目”或“依赖包已安装并应用”的结果;前者可在代码页手动运行,但须完成下节入口配置后才能由事件自动执行。
  3. 返回代码页,在 scripts/ 新建 check_dependencies,粘贴下方代码。只使用 PostgreSQL 时,删除 import pymysql 和 PyMySQL 的 version(...) 输出行;只使用 MySQL 时,删除 import psycopg 和 psycopg 的 version(...) 输出行,再保存。
  4. 选中 check_dependencies.py,点击 Run Test,在控制台核对 Python 和实际安装包的版本。
scripts/check_dependencies.pyPython
# scripts/check_dependencies.py
import sys
from importlib.metadata import version
import psycopg
import pymysql

if __name__ == "__main__":
    print("Python:", sys.version.split()[0])
    print("psycopg:", version("psycopg"))
    print("PyMySQL:", version("PyMySQL"))
    print("Dependencies ready")

完成标志 控制台显示 Python 3.13、实际安装的一个或两个包的版本,以及 Dependencies ready。

如果结果不一致

  • ModuleNotFoundError:先核对发行包与实际 import 名,再检查当前项目模块列表及安装结果;缺少依赖时补齐 Wheel、重新检查预览并安装。
  • wheel_incompatible:按实际运行目标重新准备文件。wheel_version_conflict:上传集合中同一发行包只保留一个版本。

更新或删除项目模块

模块列表仅影响当前项目。更新、删除均先显示变更预览,确认后才安装或应用。

  • 更新模块:点击“添加”,上传该模块的新版本及所需依赖,核对版本变化后安装。每个发行包只保留一个版本。
  • 删除模块:点击该行“删除”,核对清单并点击“确认删除”。其他模块仍依赖它时,先处理依赖提示。
  • 上传接受 .whl 和 Wheel 合集 ZIP;源码压缩包、安装程序、requirements.txt 不能直接导入。一次最多选择 256 个文件,总大小须小于 512 MiB。
  • 安装使用内置或已提供的 Wheel,不自动从互联网补齐缺失依赖。准备机上的 pip install 不会改变宿主机项目环境。

8. 配置正式运行入口

位置:Settings → Python → Python 模块 → 高级项目配置 → 项目运行配置。将已验证的 main(ctx) 登记为事件入口。

  1. 选择当前项目,查看 hqapp.toml 编辑区。保留系统生成的 id、package、python 与 worker 设置。
  2. 找到 [[entrypoints]] 中 id = "main" 的条目。如果已有该条目,修改它,不添加重复 id。
  3. 把 target 设置为清单实际 package 值加上 .scripts.main:main。例如 package 为 equipment_integration 时,目标为 equipment_integration.scripts.main:main;将示例包名替换为当前项目的实际 package 值。
  4. 本练习保持 class = "normal-python"、timeout_seconds = 30、max_concurrency = 1。点击“保存并应用配置”。
  5. 等待应用成功,再到运行页绑定动作。保存失败时保留错误信息并修正,不把源码已保存当成版本已生效。

完成标志 应用后,“选择脚本”应出现该入口;未出现时核对保存结果和 target。

9. 建立定时事件并绑定脚本动作

事件每 60 秒触发一次,动作选择计算脚本。绑定完成前保持事件停用。

  1. 进入配置中心 → script → 运行(Run)→ 事件(Events),点击“新增事件(New Event)”。
  2. 事件名称 / 编码填写 ScaleCheck;来源类型选择“定时循环(Timed Loop)”,循环周期填 60,最小触发间隔填 0。取消事件启用后点击“保存事件”。
  3. 选中 ScaleCheck,在“事件绑定的动作”点击“新增动作”。动作编码填写 ScaleValue,类型选择“脚本动作(Script Action)”。
  4. 在“选择脚本”选择上一步已登记的 main 脚本,保持动作启用,点击“保存动作绑定(Save Action Binding)”。
  5. 回到事件属性,勾选事件启用并保存。等待一个完整周期后打开“记录(Records)”,查看对应事件、动作、时间和执行结果。
  6. 练习完成后回到事件属性,取消启用并保存,停止周期运行。

完成标志 记录显示事件执行成功;详情中的返回结果为 {scaled: 5.0},输出包含 Scaled value: 5.0。main(ctx) 默认 raw=50,无需额外输入。

如果结果不一致

  • 没有记录:检查事件和动作是否都启用、脚本是否选中,以及 Script 服务状态。
  • 记录失败:打开该次执行的错误;若函数路径错误,修正项目配置的 target 后重新应用。
  • 有记录但不是本次结果:按事件、动作和时间核对,避免把之前的代码页运行当成定时执行。

10. 读取 Tag 并核对质量

复制本地点位 TagRef,直接替换 tags.read(...) 中的 URI 并运行。Run Test 仅从读取调用的字面量授予范围,变量不适用。对照 Tag 测试核对值与质量。tags.read_many([...]) 每次最多 256 个不重复输入,须逐项检查 ok。

  1. 准备已在 Tag 测试中显示有效数值的点位,记录其当前值与质量。
  2. 从设备目录取得当前本地站点的完整 hq://io/... TagRef;设备名称.Tag 编码与该 URI 是不同表示,不要混填。远端 Tag 不在 Run Test 的本地读取范围内。
  3. 在 scripts/ 创建 read_value,把下方 tags.read(...) 引号内的示例 URI 替换成实际 URI。
  4. 保存并通过 Run Test 运行 read_value.py,核对 Value 和 Quality。
  5. 如果现场值正在变化,以接近的采集时刻比较;先确认质量正常,再判断数值差异。
scripts/read_value.pyPython
# scripts/read_value.py
from hqcontrol import tags

def main(ctx):
    response = tags.read("hq://io/<io-domain-id>/<device-uid>/<tag-uid>")
    if not response or response.get("ok") is not True:
        raise RuntimeError("Tag read failed")
    snapshot = response["value"]
    if not tags.is_good_quality(snapshot):
        raise RuntimeError("Tag quality is not good")
    print("Value:", snapshot.get("value"), "Quality:", snapshot.get("quality"))
    return {"value": snapshot.get("value")}

if __name__ == "__main__":
    main({})

完成标志 输出目标值和 good 质量;缺值或质量异常时保留错误状态。

直接调用类型化方法

工作台也提供无需导入的 ReadTagBoolean、ReadTagInteger、ReadTagDouble、ReadTagString、ReadTagJson、ReadTagDateTime、ReadTagQuality 和对应 WriteTag 方法。Run Test 中每次读取的首个参数都应直接填写本地点位名称字面量;变量形式不授予读取范围。ReadTagQuality 将状态归为 good/bad;需要详细质量和时间戳时使用 tags.read。

PythonPython
if ReadTagQuality("<device-code>.<tag-code>") != "good":
    raise RuntimeError("Tag quality is not good")
print(ReadTagDouble("<device-code>.<tag-code>", 0.0))

calculated = ReadCalculatedTag("<calculated-tag-code>", None)
if calculated is None:
    raise RuntimeError("Calculated value is unavailable")
print(calculated)

11. 保存受控写入方法

保存为 common/tag_control.py。函数从完整读取响应取得 TagRef 和 expected 快照,写入一次并核对状态。0–100 须改为设备允许范围。Run Test 不授予写权限,仅由具备目标读写范围的正式动作调用。

  1. 在 common/ 创建 tag_control,粘贴下方代码并保存。此步骤只定义方法,不提交写入。
  2. 若要在正式动作中使用,先选择经允许的本地数值设定点,确认示例值 12.5 在设备允许范围内,并记录原值。
  3. 确认动作触发上下文授予同一目标 TagRef 的读写范围,目标允许写入且当前采集质量正常。
common/tag_control.pyPython
# common/tag_control.py
from hqcontrol import tags

def write_setpoint(read_response, value):
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= value <= 100:
        raise ValueError("Setpoint must be a number between 0 and 100")
    if not read_response or read_response.get("ok") is not True:
        raise RuntimeError("Cannot read the target Tag")
    tag_ref = read_response.get("tagRef")
    if not isinstance(tag_ref, str) or not tag_ref.startswith("hq://io/"):
        raise RuntimeError("Read response has no physical TagRef")
    snapshot = read_response["value"]
    if not tags.is_good_quality(snapshot):
        raise RuntimeError("Target quality is not good")
    result = tags.write(tag_ref, float(value), value_type="number",
                        reason="Approved setpoint update", expected=snapshot)
    if (result.get("accepted") is not True or result.get("status") != "done"
            or result.get("outcomeUncertain") is True
            or result.get("operatorVerificationRequired") is True
            or result.get("transportOnly") is True):
        raise RuntimeError("Verify command " + str(result.get("commandId")) +
                           "; status=" + str(result.get("status")) +
                           "; accepted=" + str(result.get("accepted")))
    return result

完成标志 保存完成;调用 write_setpoint 时才写入。

12. 正式动作中的受控写入参考

此入口仅供正式动作使用,Run Test 不授予写入范围。触发上下文须含目标 TagRef,并满足权限、设备和运行条件。

  1. 在 scripts/ 创建 write_setpoint,复制下方入口并完成后续替换;仅保存代码不会提交写入。
  2. 把 tags.read(...) 引号内的 URI 替换为已批准的本地设定点 TagRef,把 12.5 改为本次批准数值。
  3. 在 Run 页配置正式动作前,核对触发规则是否提供同一 TagRef 的读写范围,并排除写入再次触发本动作的循环。不要用 Code 页的 Run Test 验证写入。
  4. 正式动作执行后,在 Records 中查看命令结果,并在 Tag 测试核对后续采集值;结果不确定时按 commandId 查记录,确认前不重复写入。
scripts/write_setpoint.pyPython
# scripts/write_setpoint.py
from hqcontrol import tags
from ..common.tag_control import write_setpoint

def main(ctx):
    response = tags.read("hq://io/<io-domain-id>/<device-uid>/<tag-uid>")
    return write_setpoint(response, 12.5)

完成标志 函数仅在 accepted=true、status=done 且无不确定或人工核查标识时返回,仍须核对后续正常采集值。proxied 等其他状态保留 commandId 核查,不重复写入。

查看脚本运行状态

位置:配置中心 → script → 状态。概览与工作进程用于查看服务和执行资源;单次执行详情仍在“记录”中查看。

  • 概览:查看服务健康、工作进程摘要、触发器、连接器、调度通道和已采集的 Tag 质量。
  • 工作进程:查看就绪、忙碌、启动中、排空中等状态,以及冷启动、热命中和并发拒绝计数。
  • 服务健康不代表每个业务动作成功。遇到排队或执行失败,结合“记录”中的事件、动作、时间及错误信息排查。

运行结果与常见错误

完成示例后,控制台应输出 5.0,依赖版本与安装清单一致,读取质量为 good。写入结果需结合命令状态和后续采集值确认。

  • ModuleNotFoundError:检查发行包与实际 import 名、间接依赖、当前项目模块列表及安装结果。
  • wheel_incompatible:重新准备匹配 cp313、操作系统和架构的 Wheel;容器以容器内环境为准。
  • wheel_version_conflict:上传集合中同一发行包只保留一个版本,再重新检查安装预览。
  • 相对导入失败:检查 scripts/ 层级与点数,或改用清单实际 package 的绝对导入。
  • Tag 无值:核对授权范围、TagRef、采集质量与服务;fallback 只用于显示备用值,不代表现场数值正常。
  • 写入超时、pending_verification 或 outcome_uncertain:按 commandId 和现场回读核查,不自动重发。
Python 模块帮助中的 common 公共函数与 scripts 相对导入示例
图示 1: 同项目公共函数导入

common/ 中的函数由 scripts/ 相对导入,无需作为第三方模块安装;跨项目共享再构建 Wheel。

Settings 的 Python 模块管理界面,显示项目、运行环境和添加模块入口
图示 2: 按项目管理 Python 模块

选择项目后先核对运行环境;添加内置模块或上传 Wheel 时,都在当前项目中检查并安装。

Python 模块帮助中的网页导入与服务器目录导入步骤
图示 3: 上传与安装路径

网页导入是常用入口;服务器离线目录位于高级配置,检查目录后仍须确认安装。

再次点击图片或按 Esc 关闭