使用 ogamma Visual Logger for OPC 和 InfluxDB 存储、处理和可视化数据

作者：社区 / 产品, 用例, 开发者
2020年10月02日

本文由位于加拿大埃德蒙顿的初创公司 One-Way Automation 的软件架构师 Ravil Nugmanov 撰写。

本文介绍了一个端到端解决方案，该方案使用开源组件 InfluxDB 和 Grafana 以及 ogamma Visual Logger for OPC 构建，用于收集工业过程控制数据，以流模式分析数据，并在仪表板中可视化数据。

问题

在本演示中，我们使用了一个假设的工业泵，它通过 OPC UA 服务器发布以下值

• 输入压力
• 输出压力
• 温度
• 电源状态

在仪表板中，我们希望可视化以下信息

• 这些变量的当前值
• 图表中的历史值
• 计算输入压力和输出压力的平均值之差，并在仪表板中显示结果
• 定期检查根据上述数据计算出的压力差，并在压力差超出期望范围时生成通知，并在仪表板中显示。

解决方案

下面显示了针对此问题的提议解决方案的图表

以下部分描述了有关所用组件及其交互方式的所有详细信息。

OPC UA 数据源

作为为受监控泵提供值的数据源，我们使用 Prosys OPC UA 模拟服务器。它与其他可用的模拟服务器不同之处在于，您可以根据您的特定要求构建自定义 OPC UA 地址空间，并且可以根据选定的模式（如随机、正弦、锯齿等）自动生成变量的值。

时间序列数据库和流分析引擎

之所以选择 InfluxDB 时间序列数据库，是因为它可以高效地存储数据，并且支持实时流分析。您可以使用满足您要求的 Flux 脚本来分析数据，并即时提供结果。

将数据从 OPC UA 服务器移动到 InfluxDB

为了从 OPC UA 服务器收集数据并将其推送到 InfluxDB 时间序列数据库中，使用了 ogamma Visual Logger for OPC。它可以部署在 Windows、Linux 或 Docker 中，具有基于 Web 的配置 GUI，性能高，并且不需要太多硬件资源（用 C++ 编写）。

在仪表板中可视化信息

在本演示中，使用了另一个开源项目 Grafana。或者，也可以在 InfluxDB 中构建仪表板。

最终结果：仪表板的屏幕截图。 

最终结果：屏幕录像。

技术细节

配置 Prosys OPC UA 模拟服务器

Prosys OPC UA 模拟服务器的实例安装在 Windows 虚拟机上。在 Objects 文件夹下，创建对象和变量

从 OPC UA 客户端的角度来看，地址空间如下所示

变量配置设置在以下屏幕截图中捕获

配置 ogamma Visual Logger for OPC 以安全模式连接到服务器

首先，应在地址空间面板中创建一个新的 OPC UA 服务器节点。请注意，安全模式设置为“签名和加密”。需要在服务器端执行一些额外的步骤来配置信任。

添加 OPC UA 服务器节点后，初始连接尝试将失败。在服务器端，ogamma Visual Logger 的 OPC UA 应用程序实例证书将保存在被拒绝的证书文件夹中。从 Prosys OPC UA 模拟服务器 GUI 中，需要将其标记为受信任。这可以在切换到专家模式（菜单“选项”/“切换到专家模式”）后，通过“证书”选项卡页面完成

仅仅信任 ogamma Visual Logger 实例证书是不够的。您还需要

• 下载其 CA 证书（通过菜单“设置”/“下载证书”/“CA 证书”）并将其保存在服务器端的文件夹 C:\Users\Administrator\.prosysopc\prosys-opc-ua-simulation-server\PKI\CA\certs 中，
• 下载 CRL（通过菜单“设置”/“下载证书”/“CA CRL”）并将其保存在服务器端的文件夹 C:\Users\Administrator\.prosysopc\prosys-opc-ua-simulation-server\PKI\CA\crl 中。

注意： 证书所在的实际文件夹在您的安装中可能有所不同。因此，请使用 Prosys OPC UA 模拟服务器 GUI 打开文件夹，通过菜单“证书”/“在文件资源管理器中打开”。

当证书信任配置正确时，ogamma Visual Logger for OPC 应该能够连接到服务器。要验证这一点，请尝试通过展开地址空间面板中的节点来浏览它。

配置 ogamma Visual Logger for OPC 以记录所需的变量

从地址空间面板中选择节点，并通过单击“日志”按钮将它们添加到“已记录变量”表中。如果一切正常，则应在“状态”列中显示带有复选标记的图标。

配置 ogamma Visual Logger for OPC 以连接到 InfluxDB 实例

首先，在时间序列数据库列表中添加一个新记录（菜单“设置”/“时间序列数据库”）。将“类型”字段的值设置为“InfluxDB 2.0”，并修改字段“主机”、“端口”、“使用安全模式”和“JSON”。在最后一个字段 (JSON) 中，将选项“measurement”设置为“[TN]”，并确保正确设置选项“organization”、“bucket”、“token”。

通过单击“测试连接”按钮验证连接设置是否正确。现在，实时数据值被摄取到 InfluxDB 数据库中。

以流模式分析变量值并生成通知

对于流分析，使用了 InfluxDB Tasks 功能。通过编写 Flux 脚本来创建此功能，该脚本将每 5 秒计算一次输入压力和输出压力变量值的平均值及其差值。然后，根据此差值生成通知消息。如果差值低于指定的阈值，则消息级别设置为“严重”，否则，根据值设置为“警告”、“信息”或“正常”。然后，通知被写入特殊的系统存储桶 _monitoring，并将 _measurement 设置为 statuses。

完整的 Flux 脚本代码如下所示

import "influxdata/influxdb/monitor"

option task = {name: "Check Pump Pressure", every: 5s, offset: 1s}

currentTime = now()

join(tables: {ip: from(bucket: "ogamma")
	|> range(start: -task.every)
	|> filter(fn: (r) =>
		(r._measurement == "InputPressure" and r.unit == "Pump" and r._field == "v"))
	|> drop(columns: ["_source_timestamp"])
	|> mean(column: "_value")
	|> map(fn: (r) =>
		({
			_time: currentTime,
			_value: r._value,
			unit: r.unit,
			_measurement: "OutputPressure",
		})), op: from(bucket: "ogamma")
	|> range(start: -task.every)
	|> filter(fn: (r) =>
		(r._measurement == "OutputPressure" and r.unit == "Pump" and r._field == "v"))
	|> drop(columns: ["_source_timestamp"])
	|> mean(column: "_value")
	|> map(fn: (r) =>
		({
			_time: currentTime,
			_value: r._value,
			unit: r.unit,
			_measurement: "OutputPressure",
		}))}, on: ["_time", "_measurement"])
	|> map(fn: (r) =>
		({
			_time: r._time,
			_pd: r._value_op - r._value_ip,
			unit: r.unit_op,
			_measurement: "OutputPressure",
		}))
	|> monitor.check(
		crit: (r) =>
			(r._pd < 10.0),
		warn: (r) =>
			(r._pd < 15.0),
		info: (r) =>
			(r._pd < 25.0),
		ok: (r) =>
			(r._pd >= 25.0),
		messageFn: (r) =>
			(if r._pd < 10.0 then "Critical alert!! error: Pressure difference is too low: ${string(v: r._pd)}!" else if r._pd < 15.0 then "Warning! Pressure difference is low: ${string(v: r._pd)}!" else if r._pd < 25.0 then "Pressure difference is: ${string(v: r._pd)}!" else "ok"),
		data: {
			_check_name: "Pressure difference check",
			_check_id: "pres_diff",
			_type: "threshold",
			tags: {unit: "Pump"},
			_measurement: "statuses",
		},
	)

在 Grafana 中可视化实时和流处理的数据

Grafana 附带一个功能丰富的数据源插件，用于 InfluxDB。该插件包括自定义查询编辑器，并支持注释和查询模板。要在 Grafana 中添加 InfluxDB 作为数据源

• 通过单击顶部标题中的 Grafana 图标打开侧边菜单
• 在侧边菜单的“仪表板”链接下，您应该找到一个名为“数据源”的链接
• 单击顶部标题中的“+ 添加数据源”按钮
• 从“类型”下拉列表中选择 InfluxDB
• 确保从“查询语言”列表中选择“Flux”

每个面板都使用 Flux 脚本从 InfluxDB 查询数据。下面是每个面板的描述以及相应的 Flux 脚本代码。

状态

此面板根据布尔类型变量“Power”的值显示泵的当前状态。

from(bucket: "ogamma")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "Power" and
    r._field == "v"
  )
  |> last()
>

输入压力

在此面板中，显示输入压力变量的当前值。

from(bucket: "ogamma")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "InputPressure" and
    r._field == "v"
)

输出压力

在此面板中，显示输出压力变量的当前值。

from(bucket: "ogamma")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "OutputPressure" and
    r._field == "v"
)

相对压力

在此面板中，显示输出压力和输入压力之间计算出的差值的当前值。请注意，对于每个输入压力和输出压力变量，首先计算平均值，因此差值不等于当前值之间的差值。

from(bucket: "_monitoring")
  |> range(start: -10s, stop:now())
  |> filter(fn: (r) =>
    r._measurement == "statuses" and
    r.unit == "Pump"
  )
  |> drop (columns:["_source_timestamp", "_check_id", "_check_name", "_start", "_stop"])

泵：输入压力和输出压力

在此面板中，显示最新显示间隔内压力变量的历史值。

from(bucket: "ogamma")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "InputPressure" and
    r._field == "v"
)

泵通知

在此面板中，显示由于处理流式值而生成的通知。

from(bucket: "_monitoring")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "statuses" and
    r.unit == "Pump"
  )
  |> drop(columns: ["_source_timestamp"])

温度

在此面板中，显示温度变量的历史值。

from(bucket: "ogamma") 
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop) 
  |> filter(fn: (r) => 
    r._measurement == "Temperature" and 
    r._field == "v" 
  )

最终结果

最后，我们得到了这个在 Grafana 中工作的仪表板！