Retrieving latest value of a tag for specific device using 'latest' or 'lag' functions

[BlancoMY]

Hello everyone,

I have a use case where I collect temperature readings from multiple devices using EdgeX Foundry. Since I am using the same device profile, the resource is called Temperature for each device.

I would like to retrieve the latest value of a specific temperature based on the deviceName metadata using the latest or lag functions. However, I've encountered an issue where the meta function doesn't seem to work with these functions. It always returns the last received temperature from all devices, rather than filtering based on the deviceName metadata.

Is it possible to filter based on metadata when using the latest or lag functions? If so, could someone provide an example or guide me in the right direction?

Thank you in advance for your help! 🙏

[ngjaying]

Got it, we'll take a look. Could you please provide your SQL here?

[BlancoMY]

Thank you for your replay.
I'm using the following graph rule
{ "id": "rule1Graph", "name": "Test Condition", "graph": { "nodes": { "demo": { "type": "source", "nodeType": "edgex", "props": { "format": "json" } }, "compFilter": { "type": "operator", "nodeType": "filter", "props": { "expr": "(Temperature != nil AND meta(Temperature->deviceName)=\"modbus-device1\") OR (RoomTemperature != nil AND meta(RoomTemperature->deviceName)=\"modbus-device2\")" } }, "compFunction": { "type": "operator", "nodeType": "function", "props": { "expr": "cast( latest(Temperature) + latest(RoomTemperature), \"float\") as comp_result" } }, "pick": { "type": "operator", "nodeType": "pick", "props": { "fields": [ "comp_result" ] } }, "logOut": { "type": "sink", "nodeType": "log", "props": {} }, "edgexOut": { "type": "sink", "nodeType": "edgex", "props": { "bufferLength": 1024, "cacheLength": 1024, "cacheSaveInterval": 1024, "concurrency": 1, "connectionSelector": "edgex.redisMsgBus", "contentType": "application/json", "dataTemplate": "", "deviceName": "ekuiper", "messageType": "request", "metadata": "", "omitIfEmpty": false, "port": 6379, "profileName": "ekuiper-Profile", "protocol": "redis", "retryCount": 3, "retryInterval": 1000, "runAsync": false, "sendSingle": true, "server": "localhost", "sourceName": "comp_result", "topic": "edgex/events/device/ekuiper-service/ekuiper-Profile/ekuiper/comp_result", "topicPrefix": "", "type": "redis", "enableCache": true, "memoryCacheThreshold": 1024, "maxDiskCache": 1024000, "bufferPageSize": 256, "cleanCacheAtStop": false } } }, "topo": { "sources": [ "demo" ], "edges": { "demo": [ "compFilter" ], "compFilter": [ "compFunction" ], "compFunction": [ "pick" ], "pick": [ "logOut", "edgexOut" ] } } } }

modbus-device1 has Temperature
modbus-device2 has Temperature and RoomTemperature (both are published in one event)
I want to perform the following computation: latest(modbus-device1.Temperature) + latest(modbus-device2.RoomTemperature) but when the event of modbus-device2 is published i'm getting wrong result because it uses the Temperature of modbus-device2 instead of modbus-device1 to calculate the computation.

[BlancoMY]

You can have the same issue if you simply try to perform any computation using latest or lag of identical variable name coming from different devices.

SELECT latest(Temperature) + latest(Temperature) AS comp_result_test FROM demo WHERE (Temperature != nil AND meta(Temperature->deviceName)="modbus-device1") OR (Temperature != nil AND meta(Temperature->deviceName)="modbus-device2")

in the example above latest function retrieve the last received value of Temperature regardless of which device is coming from. The limitation is that you can use analytic functions latest and lag with only unique variable names.

[ngjaying]

To get conditional latest, use latest() OVER (WHEN a != 2). Check https://ekuiper.org/docs/en/latest/sqls/functions/analytic_functions.html#analytic-functions