Query for ranking incoming values and then assigning one of several possible categories based on the specific rank order

Fluxlang
, , ,
1

Hi Everyone!

I’m quite new to Flux/Influx and need some help some help with an appropriate query that can rank incoming values and then assign one of several possible categories based on the specific rank order.

To provide some background, we have Bluetooth tags attached to pigs in several pens at an indoor farm, where each pen has five gateways positioned around its edges. Each tag broadcasts a unique identifier (MAC address) which is recorded by each gateway that is in range, and every two seconds each gateway will upload to InfluxDB the tags it has seen during that period along with a signal strength value (RSSI) for each broadcast received. We require the flux query to identify the five gateways with the highest RSSI values for each tag during a given time interval, and then determine which pen the tag is in based on a lookup table or similar (e.g. if gateways with highest reported RSSI are A,B,C,D, and E, the tag is in pen 1, if F,G,H,I,J, the tag is in pen 2, etc.). This would then be outputted to another bucket.

The example query and screenshot below detail the RSSI value (r[“_field”] == “r”), tag id (r[“d”]), and gateway id (r[“g”]).

from(bucket: "pilot")

  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)

  |> filter(fn: (r) => r["_measurement"] == "x")

  |> filter(fn: (r) => r["_field"] == "r")

  |> filter(fn: (r) => r["d"] == "24C3")

  |> aggregateWindow(every: 300s, fn: mean, createEmpty: false)

influxTable
influxTable2811×754 159 KB

Any guidance here would be greatly appreciated, thank you!

2

Hello @somanyquestions,
Is your lookup table also in InfluxDB?
I would do something like:

from(bucket: "pilot")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "x")
  |> filter(fn: (r) => r["tag"] == "RSSI" 
//data will automatically be grouped by RSSI here
// I'm assuming you want the max value over your time period for each RSSI first
  |> max() 
// group all of your tags together to find the top 5
 |> group()
 |> top (n: 5) 
// map a new tag pen value conditionally
 |> |> map(
        fn: (r) => ({r with
            pen: if r.RSSI == "A" or r.RSSI == "B" or r.RSSI == "C" or r.RSSI == "D" or r.RSSI == "E" then
                1
            else
                2,
        }),
    )

Let me know if that helps get you started

3

Many thanks for your help here @Anaisdg.

I don’t have a lookup table in InfluxDB and to complicate things further, the conditional logic for assigning pigs to particular pens is more like the below pseudocode:

For a given device (tag r[“d”]), assign it to a Pen at each time interval (every 1 hour) as follows:

Pen 1: Gateways (tag r[“g”]) with highest RSSI values are 3DD4, 5AE0, 59CC, 7D0C, and 3E1C
Pen 2: Gateways (tag r[“g”]) with highest RSSI values are 3DD4, 5AE0, 59CC, 7CFC, and 3CC0

Can you please advise the best way to capture this within the flux query?

Thanks again,

Michael.