InfluxDB 2中如何计算自定义连续累积SOC估算值

我在InfluxDB 2中有两组时序数据：

• 仅车辆行驶时记录的电池SOC（剩余电量）值，最后一条数据为车辆停车时的SOC；
• 智能电表记录的车辆充电能量。

我需要在无车辆实际SOC测量值时估算SOC，现有Flux代码可通过reduce函数估算当前SOC，但希望得到随时间连续的估算值（即保留reduce函数的累加器至每一行），代码如下：

import "join"

car = from(bucket: "car")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "polestar2")
  |> filter(fn: (r) => r["_field"] == "batteryLevel")
  |> aggregateWindow(every: v.windowPeriod, fn: mean)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> drop(columns: ["_measurement"])

meter = from(bucket: "electricity")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "meter_wallbox")
  |> filter(fn: (r) => r["_field"] == "import_energy_active")
  |> aggregateWindow(every: v.windowPeriod, fn: spread, createEmpty: false)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> drop(columns: ["slave_id", "_measurement"])

join.full(
  left: car,
  right: meter,
  on: (l, r) => l._time == r._time,
  as: (l, r) => ({l with energy: r.import_energy_active}),
)
  |> fill(column: "energy", value: 0.0)
  |> fill(column: "batteryLevel", value: 0.0)
  |> reduce(fn: (r, accumulator) => ({
    batLevel: r.batteryLevel,
    energy: r.energy,
    estimate: if r.batteryLevel != 0.0 then r.batteryLevel else accumulator.estimate + r.energy * 1000.0
  }),
  identity: {batLevel: 0.0, energy: 0.0, estimate: 0.0})
  |> map(fn: (r) => ({ r with soc: r.estimate / 80000.0 }))
  |> drop(columns: ["batLevel", "energy", "estimate"])

内容的提问来源于stack exchange，提问作者H. Schulz