31:T52d,So far our SMART FARM has data storage provided by Firebase ([EP01](https://observablehq.com/@tomlarkworthy/livecoding-2022-06-07)), and it is actively ingesting data from real farms in Colombia using Blues Wireless cellular connection ([EP02](https://observablehq.com/@tomlarkworthy/blueswireless-2022-06-14)). We are able to send dashboards to field workers on-demand using WhatsApp ([EP03](https://observablehq.com/@tomlarkworthy/dashboards-over-whatsapp)). Last time in ([EP04](https://observablehq.com/@tomlarkworthy/firebase-to-duckdb?collection=@tomlarkworthy/livecoding)) we calculated the *Growing Degree Days* (GDD) using DuckDB SQL but it could only look back a few days before exhausting memory constraints. In this episode, we take our prototyped GDD calculations and make them work in realtime over large time ranges by precomputation. We functionify our previous dataflow based calculation logic with a flowQueue. With a function frontend, we can automate precomputing daily totals of GDD for all our sensors. With GDD precomputed over our reading history, we can utilize the lambda architecture to combine batch data with realtime data. So now we can view GDD over a month time range *and still have the latest realtime data included*. Somewhat amazingly, we have achieved this entirely within a web browser32:[] 2f:{"title":"Building a SMART FARM EP05: Lambda Architecture","cells":[{"id":0,"value":"# Building a SMART FARM EP05: Lambda Architecture\n","pinned":false,"hidden":false,"mode":"md"},{"id":365,"value":"$31","pinned":false,"hidden":false,"mode":"md"},{"id":1092,"value":"## Full episode","pinned":false,"hidden":false,"mode":"md"},{"id":343,"value":"","pinned":false,"hidden":false,"mode":"html"},{"id":643,"value":"viewof videoStartAt = Inputs.select(\n new Map([\n [\"recap\", 1],\n [\"what analytics are useful to growers?\", 45],\n [\"Tarot of the day\", 180],\n [\"Review of Growing Degrees Calculation in DuckDB SQL (EP04)\", 290],\n [\"functionification\", 1270],\n [\"lambda architecture\", 2205]\n ]),\n {\n label: \"Jump to a location of interest\"\n }\n)","pinned":false,"hidden":false,"mode":"ojs"},{"id":1933,"value":"### Realtime data over a large timeranges\n\nThis plot is a realtime view of the Growing Degree Days of some of our sensors. It would not be possible to draw this quickly if we summed all the sensor data over the time range. Instead, we precompute some daily aggregates and combine the pre-aggregated data to the latest streamed realtime data to be able to query a large timerange yet still have the latest data represented. Combining pre-aggregated 'batch' data with realtime 'streaming' data is known as a [lambda architectue](https://en.wikipedia.org/wiki/Lambda_architecture).\n","pinned":false,"hidden":false,"mode":"md"},{"id":1833,"value":"Plot.plot({\n color: {\n legend: true\n },\n y: {\n domain: [0, 24]\n },\n x: {\n type: \"time\"\n },\n marks: [\n Plot.ruleY([0]),\n Plot.lineY(historical.concat(latestData), {\n x: \"timestamp\",\n y: \"gdd\",\n stroke: \"device\"\n })\n ]\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":1958,"value":"### Historical GDD\n\nOur historical data is pulled pre-aggregated from \"aggregations/GDD\". We precompute these aggregates later on in the notebook. ","pinned":false,"hidden":false,"mode":"md"},{"id":1813,"value":"historyRaw = new Promise((resolve) =>\n db.ref(\"aggregations/GDD\").once(\"value\", (snap) => resolve(snap.val()))\n)","pinned":true,"hidden":false,"mode":"ojs"},{"id":1822,"value":"historical = {\n const data = [];\n Object.entries(historyRaw).map(([device, readings]) => [\n device,\n Object.entries(readings).forEach(([timestamp, gdd]) => {\n if (typeof gdd === \"number\")\n data.push({\n device,\n timestamp: new Date(timestamp * 1000),\n gdd\n });\n })\n ]);\n return data;\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":1963,"value":"### Realtime data\n\nOur latest data is streamed from `history/sensors/dev:864475046458393`, we are only getting data for one sensor for this simple demo.","pinned":false,"hidden":false,"mode":"md"},{"id":1847,"value":"todaySeconds = new Date(new Date().toISOString().substring(0, 10)).getTime() /\n 1000","pinned":true,"hidden":false,"mode":"ojs"},{"id":1879,"value":"latestDataRaw = Generators.observe((notify) => {\n reset;\n const dataset = (this && this.dataset) || [];\n const dataHandler = (snapshot) => {\n debugger;\n const data = snapshot.val();\n console.log(data);\n if (data.body.temperature) {\n dataset.push(data);\n notify({\n dataset,\n row: data\n });\n }\n };\n const ref = db\n .ref(`history/sensors/dev:864475046458393`)\n .orderByKey()\n .startAt(todaySeconds + \"\");\n\n ref.on(\"child_added\", dataHandler);\n\n invalidation.then(() => {\n dataset.length = 0;\n ref.off(\"child_added\", dataHandler);\n });\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":1985,"value":"We need to compute the GDD from the sensor samples. Ideally, we would reuse the SQL query used in the historical aggregation, as that was adaptive to missing data. But for simplicity we just assume each sample is 15 minutes and compute it with code. In production I would spend a little bit more time here to properly reuse the SQL query so any changes are reflected on both sides of the lambda architecture.","pinned":false,"hidden":false,"mode":"md"},{"id":1903,"value":"latestDataGdd = latestDataRaw.dataset\n .map((entry) => ({\n device: entry.device,\n timestamp: new Date(entry.when * 1000),\n temperature: entry.body.temperature\n }))\n .reduce(\n (gdd, row) =>\n gdd + (row.temperature > 5 && row.temperature < 25 ? 0.25 : 0),\n 0\n )","pinned":true,"hidden":false,"mode":"ojs"},{"id":1916,"value":"latestData = [\n {\n device: \"dev:864475046458393\",\n timestamp: new Date(todaySeconds * 1000),\n gdd: latestDataGdd\n }\n]","pinned":true,"hidden":false,"mode":"ojs"},{"id":85,"value":"## (EP01) Firebase Database\nSetup in [EP01](https://observablehq.com/@tomlarkworthy/livecoding-2022-06-07)","pinned":false,"hidden":false,"mode":"md"},{"id":61,"value":"import {\n firebase,\n viewof user,\n listen\n} with { firebaseConfig as firebaseConfig } from \"@tomlarkworthy/firebase\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":57,"value":"firebaseConfig = ({\n apiKey: \"AIzaSyA9OQ--0Sn8Gm6C1B4M-SbnxRMtKHgnHzs\",\n authDomain: \"agropatterns-iot.firebaseapp.com\",\n databaseURL: \"https://agropatterns-iot-default-rtdb.firebaseio.com\",\n projectId: \"agropatterns-iot\",\n appId: \"1:650132950601:web:115b1ba65750d2b0aaf998\",\n uiConfig: {\n // https://github.com/firebase/firebaseui-web#configuration\n signInOptions: [\"anonymous\"]\n }\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":87,"value":"db = firebase.database()","pinned":false,"hidden":false,"mode":"ojs"},{"id":176,"value":"## (EP02) Live Temperature Dashboard\n\nSetup in [EP02](https://observablehq.com/@tomlarkworthy/blueswireless-2022-06-14). This is 24 hour window of latest sensor readings. The green band represents the growth zone for the plant. Most sensors are sending data every 15 minutes which will be reflected in the dashboard automatically.","pinned":false,"hidden":false,"mode":"md"},{"id":140,"value":"viewof querySensorName = Inputs.select(Object.keys(latest), {\n label: \"Choose sensor to chart\",\n value: \"dev:864475046458393\"\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":164,"value":"dashboard = Plot.plot({\n x: { type: \"time\" },\n y: { domain: [0, 40], label: \"Temperature\" },\n color: {\n domain: [0, 40],\n scheme: \"turbo\"\n },\n marks: [\n Plot.rect([{}], { fill: \"#F4BFA6\", y1: 25, y2: 40 }),\n Plot.rect([{}], { fill: \"#A7D3A6\", y1: 5, y2: 25 }),\n Plot.rect([{}], { fill: \"#A4B5E3\", y1: 0, y2: 5 }),\n Plot.ruleY([20], { stroke: \"#0003\", strokeDasharray: [5] }),\n Plot.lineY(\n liveView.filter((d) => d.body.temperature), // Fix for unfiltering data serverside\n {\n stroke: \"#0003\",\n x: (d) => new Date(d.when * 1000),\n y: (d) => d.body.temperature\n }\n ),\n Plot.dot(\n liveView.filter((d) => d.body.temperature), // Fix for unfiltering data serverside\n {\n r: 5,\n x: (d) => new Date(d.when * 1000),\n y: (d) => d.body.temperature,\n fill: (d) => d.body.temperature\n }\n )\n ]\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":157,"value":"### streaming query (realtime updates)","pinned":false,"hidden":false,"mode":"md"},{"id":159,"value":"liveViewRaw = Generators.observe((notify) => {\n db.ref(`history/sensors/${querySensorName}`)\n .orderByKey()\n .startAt(Math.floor(Date.now() / 1000 - 60 * 60 * 24) + \"\") // reduce data returned\n .on(\"value\", (snapshot) => {\n const data = snapshot.val();\n if (data) notify(data);\n });\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":166,"value":"liveView = Object.entries(liveViewRaw).map(([k, v]) => v)","pinned":false,"hidden":false,"mode":"ojs"},{"id":372,"value":"latest = Generators.observe((notify) => {\n db.ref(`current/sensors`).once(\"value\", (snapshot) => {\n const data = snapshot.val();\n // clear synthetic data from EP01 testings\n delete data[\"sensor2\"];\n delete data[\"defaultSensorName\"];\n notify(data);\n });\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":261,"value":"import { flowQueue } from \"@tomlarkworthy/flow-queue\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":1166,"value":"## (EP04) DuckDB\n\nSetup in [EP04](https://observablehq.com/@tomlarkworthy/firebase-to-duckdb?collection=@tomlarkworthy/livecoding). We use DuckDB so we can execute SQL within a web browser using Observable awesome SQL cells. For our SMART farm we calculated \"growing degree days\" (GDD)","pinned":false,"hidden":false,"mode":"md"},{"id":1168,"value":"import { DuckDBClient } from \"@cmudig/duckdb\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":1171,"value":"duckdb = (prepareAggregation, reset, new DuckDBClient())","pinned":false,"hidden":false,"mode":"ojs"},{"id":1598,"value":"By making a second reference to a common duckdb client, we can use the first reference to populate the tables, and refresh the 2nd whenever we received new data and need to recompute our SQL cell analytic queries. ","pinned":false,"hidden":false,"mode":"md"},{"id":1244,"value":"fullduck = {\n // Refresh fullduck every insertion (causes analytics to be rerun)\n doInsertRowToDuckDB;\n return duckdb;\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":1540,"value":"### (EP04) Workflow QoL\nThe reset button allows us to reset the state and perform the data loading from scratch. *Extremely* useful during development.","pinned":false,"hidden":false,"mode":"md"},{"id":1229,"value":"mutable reset = 0","pinned":true,"hidden":false,"mode":"ojs"},{"id":2183,"value":"Inputs.button(\"reset\", { reduce: () => mutable reset++ })","pinned":false,"hidden":false,"mode":"ojs"},{"id":1482,"value":"### (EP04) Inserting Data to DuckDB\n\nFor each document sent to the flow queue we execute a SQL INSERT statement to fill DuckDB","pinned":false,"hidden":false,"mode":"md"},{"id":1180,"value":"doCreateTable = {\n reset;\n await duckdb.query(`DROP TABLE IF EXISTS readings`);\n return duckdb.query(\n `CREATE TABLE IF NOT EXISTS readings(sensor STRING, time TIMESTAMP, temperature FLOAT)`\n );\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":1186,"value":"viewof firebaseRow = (prepareAggregation,\nflowQueue({\n timeout_ms: 10000000\n}))","pinned":false,"hidden":false,"mode":"ojs"},{"id":2204,"value":"firebaseRow","pinned":true,"hidden":false,"mode":"ojs"},{"id":1194,"value":"doInsertRowToDuckDB = {\n doCreateTable;\n if (firebaseRow === null) return;\n return duckdb.query(\n `INSERT INTO readings VALUES ('${firebaseRow.device}', '${new Date(\n (firebaseRow.when || firebaseRow.tower_when) * 1000\n ).toISOString()}', ${firebaseRow.body.temperature});`\n );\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":1218,"value":"finishInsert = {\n doInsertRowToDuckDB; // Finsih processing everytime we insert a row\n viewof firebaseRow.resolve();\n}","pinned":false,"hidden":false,"mode":"ojs"},{"id":2150,"value":"### (EP04) Computing Grow Degree Days (GDD) in DuckDB","pinned":false,"hidden":false,"mode":"md"},{"id":1307,"value":"SELECT sensor, CAST(SUM(EXTRACT(epoch from growthDuration)) AS FLOAT) / (60 * 60) as growthHours\nFROM (SELECT *, CASE WHEN isGrowth THEN duration ELSE INTERVAL 0 seconds END growthDuration \n FROM (\n SELECT sensor, startTime, endTime, avgTemperature, endTime - startTime as duration, temperature > 5 AND temperature < 25 AS isGrowth \n FROM (\n SELECT sensor, time, temperature, \n AVG(temperature) OVER lookback avgTemperature,\n MIN(time) OVER lookback startTime,\n MAX(time) OVER lookback endTime\n FROM readings\n WHERE time > ${startTimeIso} AND time < ${endTimeIso}\n WINDOW lookback AS (\n ORDER BY \"time\"\n ROWS 1 PRECEDING)\n ORDER BY time\n )\n )\n ) GROUP BY sensor","pinned":true,"hidden":false,"mode":"sql","output":"aggregation","database":"var:fullduck"},{"id":1757,"value":"## EP05 Functionification of daily GDD \n\nWe took the EP04 calculations which were written in the notebook dataflow style and wrapped them in a function so we can more easily reuse the work. We created a function called `aggregate(day, sensor)` which computes the GDD for a day for a given sensor and writes it into `\"aggregations/GDD\"`. \n\nTo aid in understanding the dataflow, we open the notebook dataflow debugger:","pinned":false,"hidden":false,"mode":"md"},{"id":2181,"value":"ndd = _ndd","pinned":false,"hidden":false,"mode":"ojs"},{"id":2160,"value":"import {_ndd} from '@tomlarkworthy/ndd'","pinned":false,"hidden":false,"mode":"ojs"},{"id":2190,"value":"Inputs.button(`Aggregate ${querySensorName} on 2022-07-05`, {\n reduce: () => aggregate(\"2022-07-05\", querySensorName)\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":2251,"value":"The aggregate functions forwards to a `flowQueue`, which allows us to package up dataflow.","pinned":false,"hidden":false,"mode":"md"},{"id":1766,"value":"aggregate = (targetDay, sensor) =>\n viewof aggregationRequest.send({\n targetDay,\n sensor\n })","pinned":true,"hidden":false,"mode":"ojs"},{"id":1760,"value":"viewof aggregationRequest = (reset,\nflowQueue({\n timeout_ms: 10000000\n}))","pinned":true,"hidden":false,"mode":"ojs"},{"id":2259,"value":"The head of the queue contains the aggregation request parameters:","pinned":false,"hidden":false,"mode":"md"},{"id":1764,"value":"aggregationRequest","pinned":true,"hidden":false,"mode":"ojs"},{"id":1639,"value":"startTime = new Date(aggregationRequest.targetDay)","pinned":true,"hidden":false,"mode":"ojs"},{"id":1667,"value":"startTimeIso = startTime.toISOString()","pinned":false,"hidden":false,"mode":"ojs"},{"id":1642,"value":"endTime = new Date(startTime.getTime() + 60 * 60 * 24 * 1000)","pinned":true,"hidden":false,"mode":"ojs"},{"id":1665,"value":"endTimeIso = endTime.toISOString()","pinned":false,"hidden":false,"mode":"ojs"},{"id":2303,"value":"We first query the sensor data from `history/sensors/{sensor}` from Firebase with a \"once\" query","pinned":false,"hidden":false,"mode":"md"},{"id":1703,"value":"aggregationSensorDataRaw = new Promise((resolve) => {\n db.ref(`history/sensors/${aggregationRequest.sensor}`)\n .orderByKey()\n .startAt(startTime.getTime() / 1000 + \"\")\n .endAt(endTime.getTime() / 1000 + \"\")\n .once(\"value\", (snap) => resolve(snap.val()));\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":2354,"value":"We massage it into [tidy data](https://cran.r-project.org/web/packages/tidyr/vignettes/tidy-data.html) form in a follow up cell","pinned":false,"hidden":false,"mode":"md"},{"id":1710,"value":"aggregationSensorData = aggregationSensorDataRaw\n ? Object.entries(aggregationSensorDataRaw)\n .filter(([k, v]) => v.body.temperature)\n .map(([k, v]) => v)\n : []","pinned":true,"hidden":false,"mode":"ojs"},{"id":2263,"value":"We reset DuckDB state before inserting data","pinned":false,"hidden":false,"mode":"md"},{"id":2009,"value":"prepareAggregation = {\n aggregationSensorData;\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":2360,"value":"Our insert cell does not resolve until all data is inserted thanks to the `Promise.all`","pinned":false,"hidden":false,"mode":"md"},{"id":1718,"value":"insertIntoDb = {\n prepareAggregation;\n doCreateTable; // ensure DuckDB is reset and tables recreated\n if (aggregationSensorData.length === 0) {\n viewof firebaseRow.send(null);\n } else {\n return await Promise.all(\n aggregationSensorData.map((data) => viewof firebaseRow.send(data))\n );\n }\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":2363,"value":"We can directly retrieve the value of the SQL cell after all data has been inserted, which in our case, is computing the GDD growthHours","pinned":false,"hidden":false,"mode":"md"},{"id":1736,"value":"growthHoursAggregation = {\n insertIntoDb; // dataflow link to do after insertion\n return aggregation[0]?.growthHours || 0;\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":2373,"value":"After computing the growthHours we insert it back into Firebase at `/aggregations/GDD/`as a batch precomputation. Our lambda architecture will read this precomputed data to save time for the realtime graphs. ","pinned":false,"hidden":false,"mode":"md"},{"id":1746,"value":"writeAggregation = {\n await firebase.auth().signInAnonymously();\n return db\n .ref(\n `/aggregations/GDD/${aggregationRequest.sensor}/${\n startTime.getTime() / 1000\n }`\n )\n .set(growthHoursAggregation);\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":2376,"value":"After writing to Firebase we can consider the aggregation task complete, we return the growthHours as the 'return' value","pinned":false,"hidden":false,"mode":"md"},{"id":1777,"value":"returnResultToFlowQueue = {\n writeAggregation;\n return viewof aggregationRequest.resolve(growthHoursAggregation);\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":1773,"value":"### Orchestrating aggregate\n\nNow that aggregation has a function interface, it is more easily reused. We can create an aggregateAll function that applies aggregate to several sensors, and we can write a backfill function that calls aggregate over many days, for example.\n\nIn the future, we will want to schedule aggregate with a [cron](https://observablehq.com/@endpointservices/cron).\n\nWhat we have now is enough to demonstrate the performance gains of the lambda architecture. For a bit of fun, checkout the notebook dataflow debugger after pressing backfill button)","pinned":false,"hidden":false,"mode":"md"},{"id":2398,"value":"Inputs.button(`aggregateAll(\"2022-07-02\")`, {\n reduce: () => aggregateAll(\"2022-07-02\")\n})","pinned":false,"hidden":false,"mode":"ojs"},{"id":1797,"value":"function aggregateAll(day) {\n return Promise.all(\n [\n \"dev:864475046458393\",\n \"dev:864475046461397\",\n \"dev:864475046554399\"\n ].map((device) => aggregate(day, device))\n );\n}","pinned":true,"hidden":false,"mode":"ojs"},{"id":1942,"value":"Inputs.button(\"backfill\", {\n reduce: async () => {\n for (let i = 22; i <= 26; i++) {\n await aggregateAll(`2022-07-${(i < 10 ? \"0\" : \"\") + i}`);\n }\n }\n})","pinned":true,"hidden":false,"mode":"ojs"},{"id":1634,"value":"## Tarot of the Day\n*Will our vlog \"building a smart farm\" help with future heat waves in the united states?*\n\n

\n\n\n","pinned":false,"hidden":false,"mode":"md"},{"id":585,"value":"## Next episode","pinned":false,"hidden":false,"mode":"md"},{"id":616,"value":"upcoming","pinned":false,"hidden":false,"mode":"ojs"},{"id":453,"value":"## Dependancies","pinned":false,"hidden":false,"mode":"md"},{"id":592,"value":"import { upcoming } from \"@endpointservices/meetups\";","pinned":false,"hidden":false,"mode":"ojs"},{"id":206,"value":"import { footer } from \"@endpointservices/footer\"","pinned":false,"hidden":false,"mode":"ojs"},{"id":227,"value":"footer","pinned":false,"hidden":false,"mode":"ojs"}],"files":"$W32","stdlib":"1","titling":"auto"}