feat: moisture processor logic

server/config.json

@@ -1,17 +1,19 @@
 {
+  "water_low": false,
   "timeout": 30,
+  "check_interval": 2.5,
   "threshold": 6,
   "moisture": {
     "thresholds": {
       "lower": {
-        "0": 20, 
-        "1": 20, 
-        "2": 20
+        "sensor_0": 20, 
+        "sensor_1": 20, 
+        "sensor_2": 20
       },
       "upper": {
-        "0": 70, 
-        "1": 70, 
-        "2": 70
+        "sensor_0": 70, 
+        "sensor_1": 70, 
+        "sensor_2": 70
       }
     }
   }

server/config.json.example

@@ -1,18 +1,20 @@
 {
+  "water_low": false,
   "timeout": 30,
+  "check_interval": 2.5,
   "threshold": 6,
   "moisture": {
     "thresholds": {
       "lower": {
-        "0": 20,
-        "1": 20,
-        "2": 20
+        "sensor_0": 20, 
+        "sensor_1": 20, 
+        "sensor_2": 20
       },
       "upper": {
-        "0": 70,
-        "1": 70,
-        "2": 70
+        "sensor_0": 70, 
+        "sensor_1": 70, 
+        "sensor_2": 70
       }
     }
   }
-}
+}

server/src/config.types.ts

@@ -1,7 +1,9 @@
 export type Configuration = {
   timeout: number;
   threshold: number;
+  check_interval: number;
   moisture: Moisture;
+  water_low: boolean;
 };
 
 export type Moisture = {

server/src/cron/index.ts

@@ -24,7 +24,7 @@ export function registerCronjobs() {
   });
 }
 
-function readMoistureLevels() {
+export function readMoistureLevels() {
   return iotClient
     .get<MoistureResponse | undefined>("moisture/read")
     .catch((err: AxiosError) => {

server/src/processors/moisture.processor.ts

@@ -1,8 +1,9 @@
 import Queue from "queue";
 import { Prisma } from "@prisma/client";
 import { iotClient } from "../axios/iot.axios";
-import { MositureRow } from "../cron";
+import { MositureRow, readMoistureLevels } from "../cron";
 import { config } from "../config";
+import { emitter } from "../eventemitter";
 
 const queue = new Queue({ results: [] });
 
@@ -21,20 +22,66 @@ export function checkReadingAndEnqueue(rows: MositureRow[]) {
   });
 }
 
-function process(row: MositureRow) {
+async function process(row: MositureRow) {
   console.log("Started processing of", row.name);
-  // Enable water gate
-  // Check moisture level once per second for configured time
-  // If the moisture level does not increase a significant amount trigger water-empty alarm and clear queue
-  // If moisture level reaches upper threshold then cancel out of the checking loop
-  // Finally turn of the water gate
+  const index = row.name.replace(/[^0-9]/g, "");
+  try {
+    // Enable water gate
+    await toggleWater(index, true);
+    // Check moisture level once per second for configured time
+    Promise.race([
+      sleep(config.config.timeout * 1000),
+      timeoutCb(row.name, row.value.toNumber()),
+    ]);
+  } catch (e) {
+    console.log(`Failed to processing of ${row.name}`, e);
+  } finally {
+    // Finally turn of the water gate
+    toggleWater(index, false);
+  }
+}
+async function timeoutCb(name: string, initalRead: number): Promise<void> {
+  // If the moisture level does not increase a significant amount trigger water-empty alarm and clear queue
+  const { data } = await readMoistureLevels();
+  if (!data) {
+    throw Error(`Failed getting a moisture level reading for ${name}`);
+  }
+  if (!(name in data.data)) {
+    throw Error(
+      `Recieved invalid response for mosisture level reading for ${name}`
+    );
+  }
+  const reading = data.data[name as "sensor_0"].precentage;
+  const lowerThreshold = initalRead + config.config.threshold;
+  if (lowerThreshold >= reading) {
+    // Trigger the low water alarm
+    queue.end(Error("Water level low"));
+    // Update the configuration
+    config.config = {
+      ...config.config,
+      water_low: true,
+    };
+    // Emit the the event that the confighas been updated
+    emitter.emit("config.water_low");
+    return;
+  }
+  const upperThreshold =
+    initalRead + config.config.moisture.thresholds.upper[name as "sensor_0"];
+  if (reading >= upperThreshold) {
+    console.log(`Successfully upped the moisture level for ${name}`);
+    return;
+  }
+  // If moisture level reaches upper threshold then cancel out of the checking loop
+  await sleep(config.config.check_interval * 1000);
+  return timeoutCb(name, initalRead);
 }
-function timeoutCb() {}
 
-function toggleWater(index: number, on: boolean) {
+function toggleWater(index: string, on: boolean) {
   return iotClient
     .post(`relay/${on ? "activate" : "deactivate"}`, {
       relay: index,
     })
     .then((item) => ({ on: true }));
 }
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));