如何用Flux处理InfluxDB中暖通系统的启停状态数据？

1. 生成启停时间标注数据 #

你的数据仅在状态变化时写入，需要关联前后数据的状态差异，识别启停事件后转换成Grafana标注可用的格式。以furnace on为例，Flux查询如下：

from(bucket: "你的桶名")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "你的测量名" and r._field == "furnace on")
  |> keep(columns: ["_time", "_value"])
  // 获取上一条数据的状态和时间
  |> previous(column: "_value", as: "prev_value")
  |> previous(column: "_time", as: "prev_time")
  // 过滤出状态变化的有效行（排除无前置数据的第一条记录）
  |> filter(fn: (r) => exists r.prev_value and r._value != r.prev_value)
  // 转换为明确的启停事件，同时计算单次运行时长
  |> map(fn: (r) => ({
      _time: if r._value then r._time else r.prev_time,
      event: if r._value then "Furnace Started" else "Furnace Stopped",
      duration: if not r._value then duration(v: r._time - r.prev_time) else 0
    })
  )
  |> keep(columns: ["_time", "event", "duration"])

• previous()函数用来关联前后数据的状态与时间，以此判断状态变化。
• 通过map将状态切换转换成可读性强的启停事件，停止事件同时附带本次运行时长。
• 若需处理fan on、cooling on，只需替换_field字段，最后用union()合并多设备的事件结果即可。

2. 计算任意时间段内的总运行时长 #

利用Flux的reduce函数，在状态中同时保存上一次时间戳和累计运行时长即可解决多状态传递问题——Flux允许reduce的状态是包含多字段的记录。以furnace on为例：

from(bucket: "你的桶名")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "你的测量名" and r._field == "furnace on")
  |> keep(columns: ["_time", "_value"])
  // 补全时间范围首尾的状态，避免漏算起始已运行或结束仍运行的时长
  |> prepend(row: {_time: v.timeRangeStart, _value: false})
  |> append(row: {_time: v.timeRangeStop, _value: false})
  // 确保数据按时间顺序排列
  |> sort(columns: ["_time"])
  // 累计运行时长
  |> reduce(
      fn: (r, accumulator) => ({
          lastTime: r._time,
          totalRuntime: if accumulator.lastState then 
              accumulator.totalRuntime + duration(v: r._time - accumulator.lastTime) 
            else 
              accumulator.totalRuntime,
          lastState: r._value
        }),
      identity: {lastTime: 0, totalRuntime: 0s, lastState: false}
    )
  // 提取最终累计时长
  |> map(fn: (r) => ({total_furnace_runtime: r.totalRuntime}))

• prepend和append用来补全时间范围的首尾状态，避免因时间段开始时系统已运行、结束时仍运行导致的时长漏算。
• reduce的identity定义初始状态：lastTime记录上一条数据的时间，totalRuntime为累计时长，lastState记录上一条的运行状态。
• 每次迭代时，若上一状态为运行中（true），则计算时间差并累加到总时长里。

内容的提问来源于stack exchange，提问作者g--