IoT sensor common queries

The following scenarios illustrate common queries used to extract information from IoT sensor data:

• Calculate time in state
• Calculate time weighted average
• Calculate value between events
• Determine a state within existing values

All scenarios below use the machineProduction sample dataset provided by the InfluxDB sample package. For more information, see Sample data.

Calculate time in state

In this scenario, we look at whether a production line is running smoothly (state\=OK) and what percentage of time the production line is running smoothly or not (state\=NOK). If no points are recorded during the interval (state\=NaN), you may opt to retrieve the last state prior to the interval.

To visualize the time in state, see the Mosaic visualization.

To calculate the percentage of time a machine spends in each state

1. Import the contrib/tomhollingworth/events package.
2. Query the state field.
3. Use events.duration() to return the amount of time (in a specified unit) between each data point, and store the interval in the duration column.
4. Group columns by the status value column (in this case _value), _start, _stop, and other relevant dimensions.
5. Sum the duration column to calculate the total amount of time spent in each state.
6. Pivot the summed durations into the _value column.
7. Use map() to calculate the percentage of time spent in each state.

import "contrib/tomhollingworth/events"
from(bucket: "machine")
    |> range(start: 2021-08-01T00:00:00Z, stop: 2021-08-02T00:30:00Z)
    |> filter(fn: (r) => r["_measurement"] == "machinery")
    |> filter(fn: (r) => r["_field"] == "state")
    |> events.duration(unit: 1h, columnName: "duration")
    |> group(columns: ["_value", "_start", "_stop", "station_id"])
    |> sum(column: "duration")
    |> pivot(rowKey: ["_stop"], columnKey: ["_value"], valueColumn: "duration")
    |> map(
        fn: (r) => {
            totalTime = float(v: r.NOK + r.OK)
            return {r with NOK: float(v: r.NOK) / totalTime * 100.0, OK: float(v: r.OK) / totalTime * 100.0}
        },
    )

The query above focuses on a specific time range of state changes reported in the production line.

• range() defines the time range to query.
• filter() defines the field (state) and measurement (machinery) to filter by.
• events.duration() calculates the time between points.
• group() regroups the data by the field value, so points with OK and NOK field values are grouped into separate tables.
• sum() returns the sum of durations spent in each state.

The output of the query at this point is:

pivot() creates columns for each unique value in the _value column, and then assigns the associated duration as the column value. The output of the pivot operation is:

Given the output above, map() does the following:

1. Adds the NOK and OK values to calculate totalTime.
2. Divides NOK by totalTime, and then multiplies the quotient by 100.
3. Divides OK by totalTime, and then multiplies the quotient by 100.

This returns:

The result shows that 88.66% of time production is in the OK state, and that 11.34% of time, production is in the NOK state.

Mosaic visualization

The mosaic visualization displays state changes over time. In this example, the mosaic visualization displays different colored tiles based on the state field.

from(bucket: "machine")
    |> range(start: 2021-08-01T00:00:00Z, stop: 2021-08-02T00:30:00Z)
    |> filter(fn: (r) => r._measurement == "machinery")
    |> filter(fn: (r) => r._field == "state")
    |> aggregateWindow(every: v.windowPeriod, fn: last, createEmpty: false)

When visualizing data, it is possible to have more data points than available pixels. To divide data into time windows that span a single pixel, use aggregateWindow with the every parameter set to v.windowPeriod. Use last as the aggregate fn to return the last value in each time window. Set createEmpty to false so results won’t include empty time windows.

Calculate time weighted average

To calculate the time-weighted average of data points, use the timeWeightedAvg() function.

The example below queries the oil_temp field in the machinery measurement. The timeWeightedAvg() function returns the time-weighted average of oil temperatures based on 5 second intervals.

from(bucket: "machine")
    |> range(start: 2021-08-01T00:00:00Z, stop: 2021-08-01T00:00:30Z)
    |> filter(fn: (r) => r._measurement == "machinery" and r._field == "oil_temp")
    |> timeWeightedAvg(unit: 5s)

Output data

Calculate value between events

Calculate the value between events by getting the average value during a specific time range.

The following scenario queries data starting when four production lines start and end. The following query calculates the average oil temperature for each grinding station during that period.

batchStart = 2021-08-01T00:00:00Z
batchStop = 2021-08-01T00:00:20Z
from(bucket: "machine")
    |> range(start: batchStart, stop: batchStop)
    |> filter(fn: (r) => r._measurement == "machinery" and r._field == "oil_temp")
    |> mean()

Output

Determine a state with existing values

Use multiple existing values to determine a state. The following example calculates a state based on the difference between the pressure and pressure-target fields in the machine-production sample data. To determine a state by comparing existing fields:

1. Query the fields to compare (in this case, pressure and pressure_target).
2. (Optional) Use aggregateWindow() to window data into time-based windows and apply an aggregate function (like mean()) to return values that represent larger windows of time.
3. Use pivot() to shift field values into columns.
4. Use map() to compare or operate on the different field column values.
5. Use map() to assign a status (in this case, needsMaintenance based on the relationship of the field column values.

import "math"
from(bucket: "machine")
    |> range(start: 2021-08-01T00:00:00Z, stop: 2021-08-02T00:00:00Z)
    |> filter(fn: (r) => r["_measurement"] == "machinery")
    |> filter(fn: (r) => r["_field"] == "pressure" or r["_field"] == "pressure_target")
    |> aggregateWindow(every: 12h, fn: mean)
    |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
    |> map(fn: (r) => ({ r with pressureDiff: r.pressure - r.pressure_target }))
    |> map(fn: (r) => ({ r with needsMaintenance: if math.abs(x: r.pressureDiff) >= 15.0 then true else false }))

Output

The table reveals that the pressureDiff value -19.636860591766734 from station g4 and -19.049240090497733 from station g3 are higher than 15, therefore there is a change in state that marks the needMaintenance value as “true” and would require that station to need work to turn that value back to false.