how to manipulate incoming data stream?

[barrymac]

Well after many iterations and some very exciting live coding in production while trying to prevent the boat from randomly gybing, this is my damping algorithm which will discard any bad data that frequently comes through from old instruments and uses the average value from a set, with a configurable size.

The damping set and the tolerance for difference from the last average are both configurable. Hopefully this helps someone else!

Code:

[
   {
       "id": "48e91ee0.6e36b",
       "type": "tab",
       "label": "Clean AWA",
       "disabled": false,
       "info": ""
   },
   {
       "id": "cf421cb3.e95bc",
       "type": "inject",
       "z": "48e91ee0.6e36b",
       "name": "",
       "props": [
           {
               "p": "payload"
           },
           {
               "p": "topic",
               "vt": "str"
           }
       ],
       "repeat": "",
       "crontab": "",
       "once": false,
       "onceDelay": "1",
       "topic": "environment.wind.angleApparent",
       "payload": "-3.14",
       "payloadType": "num",
       "x": 170,
       "y": 80,
       "wires": [
           [
               "4b11a19a.2aae8"
           ]
       ]
   },
   {
       "id": "4b11a19a.2aae8",
       "type": "function",
       "z": "48e91ee0.6e36b",
       "name": "filter AWA more than 30% different",
       "func": "let threshold = 17;\nlet maxValue = 3.15;\n\nlet minValue = -3.15;\nlet dampingSet = 5;\nlet modifyMsg = flow.get(\"modifyMsg\");\n\nreturn modifyMsg(msg,threshold,maxValue,minValue,dampingSet);\n",
       "outputs": 2,
       "noerr": 0,
       "initialize": "let modifyMsg = function modifyMsg(message = msg,\n                                   thresholdVal = threshold,\n                                   max = maxValue,\n                                   min = minValue,\n                                   dampingSet = 5) {\n\n    let currentValue = message.payload;\n    let lastKnownGoodValue = flow.get(\"lastKnownGoodValue\");\n    let prevAvgValue = flow.get(\"avgValue\");\n\n    // dampingSet; //number of elements to use for average value\n    // node.warn(\"damping array\"+JSON.stringify(flow.get(\"dampingAry\")));\n    if (flow.get(\"dampingAry\") == null) {\n        node.warn(\"initialising damping array\");\n        let damping = Array.from({length: dampingSet}, () => ({value: 0, update: 0, init: 0,}))\n        damping[0].update = 1\n        let dampingElement = 0;\n        flow.set(\"dampingAry\", damping);\n    }\n\n    let differenceInRads = (Math.abs(lastKnownGoodValue) - Math.abs(currentValue));\n    let percent = ((differenceInRads) / (2 * Math.PI)) * 100;\n\n    if (currentValue != null && !isNaN(currentValue) && currentValue != 0) {\n        //if value is not blank then proceed, otherwise use previous value\n        if ((Math.abs(currentValue) > max) || (Math.abs(currentValue) < min)) {\n            //if value is a crazy number, then use last average value\n            node.warn(JSON.stringify(currentValue* 180/Math.PI) + \" impossible value: \" + JSON.stringify(currentValue* 180/Math.PI));\n            message.payload = prevAvgValue\n            // node.warn(\"corrected message value: \" + JSON.stringify(message.payload));\n            return message;\n        } else if ((percent > thresholdVal)) {\n            //if value is too different from the average in the damping array, then use last known good value\n            // message.payload = lastKnownGoodValue\n            message.payload = prevAvgValue\n            // node.warn(JSON.stringify(percent) + \" % diff \" + differenceInRads + \" in rads  \" +\n            //     \"between \" + JSON.stringify(lastKnownGoodValue) + \" and \" + JSON.stringify(currentValue)\n            // + \" too high using previous average value: \" + JSON.stringify(prevAvgValue));\n            // // node.warn(differenceInDeg+\"  in degs between \" + JSON.stringify(lastKnownGoodValue))\n            node.warn(\"corrected \" + JSON.stringify(currentValue* 180/Math.PI)\n                + \" to: \"\n                + JSON.stringify(message.payload* 180/Math.PI)\n                + \" flow.get(\\\"avgValue\\\"): \"\n                + JSON.stringify(flow.get(\"avgValue\")* 180/Math.PI)\n                + \" lastKnownGoodValue: \"\n                + JSON.stringify(flow.get(\"lastKnownGoodValue\")* 180/Math.PI)\n                // + \" dampingAry: \"\n                // + JSON.stringify(flow.get(\"dampingAry\"))\n            );\n            return message;\n        } else {\n            //if value seems ok then update last known good value, damping array and average value\n            flow.set(\"lastKnownGoodValue\", currentValue);\n            updateDampingArray(currentValue);\n        }\n        //value seems ok so pass it on unmodified\n        // node.warn(\"uncorrected message value: \" + JSON.stringify(message.payload));\n        return message;\n    } else {\n        flow.set(\"avgValue\", prevAvgValue);\n        // message.payload = lastKnownGoodValue\n        message.payload = prevAvgValue\n        // node.warn(\"currentValue null, using last value: \" + JSON.stringify(lastKnownGoodValue* 180/Math.PI));\n        node.warn(\"currentValue null, using last avg: \" + JSON.stringify(prevAvgValue* 180/Math.PI));\n        node.warn(\"average set to: \" + JSON.stringify(prevAvgValue* 180/Math.PI));\n    }\n}\n\nfunction updateDampingArray(currentValue) {\n    //todo convert negatives to 180-360 for averaging and back again\n    let ary = flow.get(\"dampingAry\");\n    ary.forEach((element, index, array) => {\n        if (element.update == 1) {\n            dampingElement = index;\n            element.update = 0\n        }\n    });\n    if (currentValue<0){\n        ary[dampingElement].value = currentValue +(2*Math.PI);\n    } else {\n        ary[dampingElement].value = currentValue;\n    }\n\n    ary[dampingElement].update = 0;\n    if (ary[dampingElement + 1] != null) {\n        ary[dampingElement + 1].update = 1;\n    } else {\n        ary[0].update = 1;\n    }\n\n    let avgVal = ary.reduce(function (sum, element) {\n        return sum + parseFloat(element.value);\n    }, 0) / ary.length;\n\n    if (avgVal != null && !isNaN(avgVal) && avgVal != 0) {\n        //defensively update average\n        if (avgVal>Math.PI) {\n            flow.set(\"avgValue\", avgVal-(2*Math.PI));\n        } else {\n            flow.set(\"avgValue\", avgVal);\n        }\n\n        // if (avgVal != prevAvgValue) {\n        //     node.warn(\"update average from \" + prevAvgValue + \" to:  \" + JSON.stringify(avgVal));\n        // }\n    }\n\n    flow.set(\"dampingAry\", ary);\n}\n\nflow.set(\"modifyMsg\", modifyMsg);\n",
       "finalize": "",
       "libs": [],
       "x": 460,
       "y": 140,
       "wires": [
           [
               "670bd0ea.8ea67"
           ],
           []
       ]
   },
   {
       "id": "58e0a1ba.0ddcb",
       "type": "signalk-input-handler",
       "z": "48e91ee0.6e36b",
       "name": "AWA input handler",
       "context": "vessels.self",
       "path": "environment.wind.angleApparent",
       "source": "",
       "x": 130,
       "y": 140,
       "wires": [
           [
               "4b11a19a.2aae8"
           ]
       ]
   },
   {
       "id": "670bd0ea.8ea67",
       "type": "signalk-input-handler-next",
       "z": "48e91ee0.6e36b",
       "name": "back to server",
       "x": 760,
       "y": 140,
       "wires": []
   },
   {
       "id": "deeb0c7c.106ce",
       "type": "inject",
       "z": "48e91ee0.6e36b",
       "name": "",
       "props": [
           {
               "p": "payload"
           },
           {
               "p": "topic",
               "vt": "str"
           }
       ],
       "repeat": "",
       "crontab": "",
       "once": false,
       "onceDelay": "1",
       "topic": "environment.wind.angleApparent",
       "payload": "2.94",
       "payloadType": "num",
       "x": 170,
       "y": 40,
       "wires": [
           [
               "4b11a19a.2aae8"
           ]
       ]
   }
]