IoT sensor common queries

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

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:

_value duration
NOK 22
_value duration
OK 172

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:

22 172

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:

11.34020618556701 88.65979381443299

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
stationID _start _stop _value
g1 2021-08-01T01:00:00.000Z 2021-08-01T00:00:30.000Z 40.25396118491921
g2 2021-08-01T01:00:00.000Z 2021-08-01T00:00:30.000Z 40.6
g3 2021-08-01T01:00:00.000Z 2021-08-01T00:00:30.000Z 41.384505595567866
g4 2021-08-01T01:00:00.000Z 2021-08-01T00:00:30.000Z 41.26735518634935

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()
stationID _start _stop _value
g1 2021-08-01T01:00:00.000Z 2021-08-02T00:00:00.000Z 40
g2 2021-08-01T01:00:00.000Z 2021-08-02T00:00:00.000Z 40.6
g3 2021-08-01T01:00:00.000Z 2021-08-02T00:00:00.000Z 41.379999999999995
g4 2021-08-01T01:00:00.000Z 2021-08-02T00:00:00.000Z 41.2

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 }))
_time needsMaintenance pressure pressure_target pressureDiff stationID
2021-08-01T12:00:00.000Z false 101.83929080014092 104.37786394078252 -2.5385731406416028 g1
2021-08-02T00:00:00.000Z false 96.04368008245874 102.27698650674662 -6.233306424287889 g1
_time needsMaintenance pressure pressure_target pressureDiff stationID
2021-08-01T12:00:00.000Z false 101.62490431541765 104.83915260886623 -3.214248293448577 g2
2021-08-02T00:00:00.000Z false 94.52039415465273 105.90869375273046 -11.388299598077722 g2
_time needsMaintenance pressure pressure_target pressureDiff stationID
2021-08-01T12:00:00.000Z false 92.23774168403503 104.81867444768653 -12.580932763651504 g3
2021-08-02T00:00:00.000Z true 89.20867846153847 108.2579185520362 -19.049240090497733 g3
_time needsMaintenance pressure pressure_target pressureDiff stationID
2021-08-01T12:00:00.000Z false 94.40834093349847 107.6827757125155 -13.274434779017028 g4
2021-08-02T00:00:00.000Z true 88.61785638936534 108.25471698113208 -19.636860591766734 g4

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.

Was this page helpful?

Thank you for your feedback!

Introducing InfluxDB Clustered

A highly available InfluxDB 3.0 cluster on your own infrastructure.

InfluxDB Clustered is a highly available InfluxDB 3.0 cluster built for high write and query workloads on your own infrastructure.

InfluxDB Clustered is currently in limited availability and is only available to a limited group of InfluxData customers. If interested in being part of the limited access group, please contact the InfluxData Sales team.

Learn more
Contact InfluxData Sales

The future of Flux

Flux is going into maintenance mode. You can continue using it as you currently are without any changes to your code.

Flux is going into maintenance mode and will not be supported in InfluxDB 3.0. This was a decision based on the broad demand for SQL and the continued growth and adoption of InfluxQL. We are continuing to support Flux for users in 1.x and 2.x so you can continue using it with no changes to your code. If you are interested in transitioning to InfluxDB 3.0 and want to future-proof your code, we suggest using InfluxQL.

For information about the future of Flux, see the following:

InfluxDB Cloud powered by TSM