THELIMA :Clarity in every drop

#include <Arduino_RouterBridge.h>
#include <OneWire.h>
#include <DallasTemperature.h>
#include <math.h>

// ---------- Pin Assignments ----------
#define turbidityPin A0
#define tdsPin       A1
#define phPin        A2
#define ONE_WIRE_BUS D8  // DS18B20

OneWire oneWire(ONE_WIRE_BUS);
DallasTemperature sensors(&oneWire);

// ---------- 3.3V System & 12-Bit ADC Settings ----------
#define VREF 3.3
#define ADC_MAX 4095.0

// ---------- Auto-calibration ----------
// With AUTO_CALIBRATE on, hold every probe in clean tap water when the app
// starts. The first CAL_SAMPLES readings are averaged and used as the
// reference point: that water becomes pH 7.0 and 0 NTU. Set to false and fill
// in the constants below by hand once you have proper buffer solutions.
#define AUTO_CALIBRATE true
const int CAL_SAMPLES = 20;

// Turbidity: NTU from the DFRobot/SEN0189 response curve
//   NTU = -1120.4*V^2 + 5742.3*V - 4352.9
// defined for the sensor's native 0-4.5 V swing, so the measured voltage is
// scaled up to that range using the clean-water reference.
float turbidityVClear = 2435 * (VREF / ADC_MAX);  // overwritten by auto-cal
const float TURBIDITY_V_CURVE_CLEAR = 4.2;        // curve's 0 NTU voltage
const float TURBIDITY_NTU_MAX = 3000.0;

// pH: probe modules are near-linear with a negative slope. -5.56 pH per volt
// is the standard PH-4502C figure; refine it with a pH 4 buffer if you have
// one (slope = 3.0 / (voltage_pH4 - voltage_pH7)).
float voltage_pH7 = 0.149;             // overwritten by auto-cal
const float PH_SLOPE = -5.56;          // pH per volt

// TDS: multiply if your meter reads consistently low/high. If you know the
// ppm of the water the buoy starts in (a TDS pen reading, or ~50 ppm for
// typical fresh water), set TDS_REFERENCE_PPM and the scale is solved for it
// at startup. 0 disables the one-point calibration.
float tdsScale = 1.0;
const float TDS_REFERENCE_PPM = 0.0;
const float TDS_PPM_MAX = 2000.0;

// Turbidity reads a few NTU in clean water because of stray light and probe
// tolerance. The NTU computed during calibration is treated as the zero, so
// the demo water starts at 0 instead of a random offset.
float turbidityNtuOffset = 0.0;

// Plausibility band for fresh water. Outside it the probe is almost certainly
// out of the water or unplugged, and the reading is reported but flagged.
const float PH_PLAUSIBLE_MIN = 3.0;
const float PH_PLAUSIBLE_MAX = 11.0;

// Late pH re-baseline. If the probe was still powering up when setup()
// calibrated, pH 7 stays pinned to the default voltage and every reading
// lands outside the plausible band. Once the probe has held a steady voltage
// in the water for PH_REBASELINE_SAMPLES readings, that voltage is taken as
// pH 7.00 - the same one-point calibration setup() does, only later.
#define PH_AUTO_REBASELINE true
const int PH_REBASELINE_SAMPLES = 5;
const float PH_STEADY_TOLERANCE = 0.01;
int phSteadyCount = 0;
float phLastVoltage = 0;
bool phRebaselined = false;

// Every analog reading is a median of MEDIAN_SAMPLES to kill ADC spikes, then
// smoothed so the value shown and sent to the model does not jitter.
const int MEDIAN_SAMPLES = 9;
const float SMOOTHING = 0.25;   // 0 = frozen, 1 = no smoothing
float smoothTurb = 0, smoothTds = 0, smoothPh = 0;
bool smoothingPrimed = false;

// ---------- Probe warm-up ----------
// The analog probe modules need a few seconds after power-up before their
// output settles; sampling a baseline before that pins the calibration to a
// meaningless voltage. Calibration waits for the readings to rise above the
// minimum plausible ADC value and to stop drifting.
const unsigned long WARMUP_MS = 5000;
const unsigned long SETTLE_TIMEOUT_MS = 30000;
const int TURBIDITY_ADC_MIN = 500;   // below this the probe is not powered
const int PH_ADC_MIN = 200;          // below this the probe is not powered
const float SETTLE_TOLERANCE = 0.02; // 2% drift between consecutive averages

// Sampling interval. Must match the frequency of your impulse (1000 ms = 1 Hz).
const unsigned long SAMPLE_INTERVAL_MS = 1000;
unsigned long previousMillis = 0;

// ---------- SIM800L v2 SMS alerts ----------
// SIM800L TX -> D0 (RX), SIM800L RX -> D1 (TX)  == Serial1 on the UNO Q.
#define SIM_SERIAL Serial1
#define SIM_BAUD   9600
const char *SMS_RECIPIENT = "+918891206899";
const char *SOURCE_NAME   = "Ambedkar colony Pump house";
const char *DUMMY_DATETIME = "05-09-2026 14:30";   // placeholder until an RTC/NTP is added

// Anomaly score -> risk 0..10 (log scale):
//   <=0.1 -> 0, 0.3 -> 2, 1 -> 3, 3 -> 5, 10 -> 7, 30 -> 8, 100 -> 10
// Thresholds are the anomaly values at which 3.33*(log10(a)+1) rounds up to
// the next integer, so no libm call is needed on the MCU.
int riskOutOf10(float anomaly) {
  static const float kRiskThresholds[10] = {
    0.1413f, 0.2822f, 0.5634f, 1.1249f, 2.2461f,
    4.4842f, 8.9532f, 17.876f, 35.696f, 71.267f
  };
  int risk = 0;
  for (int i = 0; i < 10; i++) {
    if (anomaly >= kRiskThresholds[i]) risk = i + 1;
  }
  return risk;
}
// A routine report goes out at most every REPORT_INTERVAL_MS; an ANOMALY
// verdict is sent right away but never more often than ALERT_INTERVAL_MS.
// Demo mode: 0 = send on every model window (~28 s). For deployment use e.g.
// 30UL*60UL*1000UL (30 min) and 5UL*60UL*1000UL (5 min).
const unsigned long REPORT_INTERVAL_MS = 0;
const unsigned long ALERT_INTERVAL_MS  = 0;
unsigned long lastReportMillis = 0;
unsigned long lastAlertMillis  = 0;
bool firstResult = true;

// Latest readings, kept so the SMS can include them.
float lastTemp = 0, lastTurb = 0, lastTds = 0, lastPh = 0;

void simFlush(unsigned long ms) {
  unsigned long t = millis();
  while (millis() - t < ms) {
    while (SIM_SERIAL.available()) Serial.write(SIM_SERIAL.read());
  }
}

void simCommand(const char *cmd, unsigned long waitMs) {
  SIM_SERIAL.println(cmd);
  simFlush(waitMs);
}

void simSetup() {
  SIM_SERIAL.begin(SIM_BAUD);
  simFlush(1000);
  simCommand("AT", 500);
  simCommand("ATE0", 500);       // echo off
  simCommand("AT+CMGF=1", 500);  // text mode
  simCommand("AT+CSCS=\"GSM\"", 500);
  simCommand("AT+CREG?", 500);   // network registration (want +CREG: 0,1)
  simCommand("AT+CSQ", 500);     // signal quality
}

// Read from the modem until `token` shows up or `ms` elapses. Everything read
// is echoed to the Serial Monitor. Returns true if the token was seen.
bool simWaitFor(const char *token, unsigned long ms) {
  String buf;
  unsigned long t = millis();
  while (millis() - t < ms) {
    while (SIM_SERIAL.available()) {
      char c = SIM_SERIAL.read();
      Serial.write(c);
      buf += c;
      if (buf.indexOf(token) >= 0) return true;
      if (buf.indexOf("ERROR") >= 0) return false;
      if (buf.length() > 200) buf = buf.substring(100);
    }
  }
  return false;
}

bool sendSmsOnce(const String &text) {
  while (SIM_SERIAL.available()) SIM_SERIAL.read();   // drop stale output

  SIM_SERIAL.println("AT");
  if (!simWaitFor("OK", 1000)) { Serial.println("[SMS] modem not answering"); return false; }

  SIM_SERIAL.println("AT+CMGF=1");
  simWaitFor("OK", 1000);

  SIM_SERIAL.print("AT+CMGS=\"");
  SIM_SERIAL.print(SMS_RECIPIENT);
  SIM_SERIAL.println("\"");
  if (!simWaitFor(">", 5000)) {
    Serial.println("[SMS] no '>' prompt");
    SIM_SERIAL.write(27);                              // ESC aborts a half-open CMGS
    return false;
  }

  SIM_SERIAL.print(text);
  SIM_SERIAL.write(26);                                // Ctrl-Z sends
  if (!simWaitFor("+CMGS", 30000)) {                   // network can take 10-20 s
    Serial.println("[SMS] not accepted by network");
    return false;
  }
  simWaitFor("OK", 2000);
  return true;
}

bool sendSms(const String &text) {
  for (int attempt = 1; attempt <= 3; attempt++) {
    Serial.print("[SMS] sending, attempt "); Serial.println(attempt);
    if (sendSmsOnce(text)) { Serial.println("[SMS] sent"); return true; }
    delay(3000);
  }
  Serial.println("[SMS] FAILED after 3 attempts");
  return false;
}

float readAveraged(int pin) {
  long total = 0;
  for (int i = 0; i < CAL_SAMPLES; i++) {
    total += analogRead(pin);
    delay(20);
  }
  return (float)total / CAL_SAMPLES;
}

// Median of MEDIAN_SAMPLES consecutive reads - one stray conversion cannot
// move the result the way it moves an average.
int readMedian(int pin) {
  int v[MEDIAN_SAMPLES];
  for (int i = 0; i < MEDIAN_SAMPLES; i++) {
    v[i] = analogRead(pin);
    delay(2);
  }
  for (int i = 1; i < MEDIAN_SAMPLES; i++) {
    int key = v[i], j = i - 1;
    while (j >= 0 && v[j] > key) { v[j + 1] = v[j]; j--; }
    v[j + 1] = key;
  }
  return v[MEDIAN_SAMPLES / 2];
}

float smooth(float previous, float current) {
  return smoothingPrimed ? previous + SMOOTHING * (current - previous) : current;
}

float turbidityFromVoltage(float voltage) {
  float vScaled = voltage * (TURBIDITY_V_CURVE_CLEAR / turbidityVClear);
  float ntu = -1120.4 * vScaled * vScaled + 5742.3 * vScaled - 4352.9;
  return ntu;
}

float tdsFromVoltage(float voltage, float temperature) {
  float compVoltage = voltage / (1.0 + 0.02 * (temperature - 25.0));
  return (133.42 * compVoltage * compVoltage * compVoltage
          - 255.86 * compVoltage * compVoltage
          + 857.39 * compVoltage) * 0.5;
}

// Average `pin` once the probe is powered and its output has stopped drifting.
// Returns 0 if that never happens within SETTLE_TIMEOUT_MS.
float readSettledBaseline(int pin, int minADC, const char *name) {
  unsigned long start = millis();
  float previous = readAveraged(pin);

  while (millis() - start < SETTLE_TIMEOUT_MS) {
    float current = readAveraged(pin);
    if (current >= minADC && previous >= minADC) {
      float delta = current - previous;
      if (delta < 0) delta = -delta;
      float drift = delta / current;
      if (drift <= SETTLE_TOLERANCE) return current;
    }
    previous = current;
    delay(200);
  }

  Serial.print("  WARNING: "); Serial.print(name);
  Serial.println(" never settled - check the wiring. Keeping default calibration.");
  return 0;
}

void setup() {
  Serial.begin(9600);
  analogReadResolution(12);

  sensors.begin();
  sensors.setWaitForConversion(true);
  sensors.setResolution(10);

  Bridge.begin();
  // Linux/Python calls this back with the model output so it shows up in the
  // Serial Monitor.
  Bridge.provide("show_result", showResult);

  Serial.println("Water Quality AI - MCU ready");
  simSetup();

  if (AUTO_CALIBRATE) {
    Serial.println("Waiting for the probes to warm up...");
    delay(WARMUP_MS);

    Serial.println("Calibrating - keep the probes in clean water...");

    float turbBaseline = readSettledBaseline(turbidityPin, TURBIDITY_ADC_MIN, "turbidity");
    if (turbBaseline > 0) turbidityVClear = turbBaseline * (VREF / ADC_MAX);

    float phBaseline = readSettledBaseline(phPin, PH_ADC_MIN, "pH");
    if (phBaseline > 0) voltage_pH7 = phBaseline * (VREF / ADC_MAX);

    // Whatever NTU the curve produces for the calibration water becomes the
    // zero, so the buoy starts the demo at 0 NTU rather than a few NTU of
    // stray-light offset.
    if (turbBaseline > 0) {
      turbidityNtuOffset = turbidityFromVoltage(turbidityVClear);
    }

    // One-point TDS calibration against the known ppm of the demo water.
    if (TDS_REFERENCE_PPM > 0) {
      sensors.requestTemperatures();
      float t = sensors.getTempCByIndex(0);
      if (t == DEVICE_DISCONNECTED_C || t < -50.0) t = 25.0;
      float raw = tdsFromVoltage(readAveraged(tdsPin) * (VREF / ADC_MAX), t);
      if (raw > 1.0) tdsScale = TDS_REFERENCE_PPM / raw;
    }

    Serial.print("  turbidity 0 NTU at ADC "); Serial.print(turbBaseline, 0);
    Serial.print(" ("); Serial.print(turbidityVClear, 3); Serial.print(" V, offset ");
    Serial.print(turbidityNtuOffset, 1); Serial.println(" NTU)");
    Serial.print("  pH 7.00 at ADC ");         Serial.print(phBaseline, 0);
    Serial.print(" ("); Serial.print(voltage_pH7, 3); Serial.println(" V)");
    Serial.print("  TDS scale ");              Serial.println(tdsScale, 3);
    Serial.println("Calibration done.");
  }
}

void loop() {
  unsigned long now = millis();
  if (now - previousMillis < SAMPLE_INTERVAL_MS) {
    return;
  }
  previousMillis = now;

  // ---------------- 1. Temperature (DS18B20) ----------------
  sensors.requestTemperatures();
  float temperature = sensors.getTempCByIndex(0);
  bool tempOk = true;
  if (temperature == DEVICE_DISCONNECTED_C || temperature < -50.0) {
    temperature = 25.0;
    tempOk = false;
  }

  // ---------------- 2. Turbidity (NTU) ----------------
  int turbADC = readMedian(turbidityPin);
  float turbVoltage = turbADC * (VREF / ADC_MAX);
  float turbidity = turbidityFromVoltage(turbVoltage) - turbidityNtuOffset;
  turbidity = constrain(turbidity, 0.0, TURBIDITY_NTU_MAX);
  smoothTurb = smooth(smoothTurb, turbidity);
  turbidity = smoothTurb;

  // ---------------- 3. TDS ----------------
  int tdsADC = readMedian(tdsPin);
  float tdsVoltage = tdsADC * (VREF / ADC_MAX);
  float tdsValue = tdsFromVoltage(tdsVoltage, temperature) * tdsScale;
  tdsValue = constrain(tdsValue, 0.0, TDS_PPM_MAX);
  smoothTds = smooth(smoothTds, tdsValue);
  tdsValue = smoothTds;

  // ---------------- 4. pH ----------------
  int phADC = readMedian(phPin);
  float phVoltage = phADC * (VREF / ADC_MAX);

  if (PH_AUTO_REBASELINE && !phRebaselined && phADC >= PH_ADC_MIN) {
    float uncalibrated = 7.0 + PH_SLOPE * (phVoltage - voltage_pH7);
    float drift = phVoltage - phLastVoltage;
    if (drift < 0) drift = -drift;
    bool steady = phLastVoltage > 0 && drift / phVoltage <= PH_STEADY_TOLERANCE;
    bool implausible = uncalibrated < PH_PLAUSIBLE_MIN || uncalibrated > PH_PLAUSIBLE_MAX;

    if (steady && implausible) {
      if (++phSteadyCount >= PH_REBASELINE_SAMPLES) {
        voltage_pH7 = phVoltage;
        phRebaselined = true;
        smoothingPrimed = false;
        Serial.print("pH re-calibrated: 7.00 is now ");
        Serial.print(voltage_pH7, 3);
        Serial.println(" V");
      }
    } else {
      phSteadyCount = 0;
    }
    phLastVoltage = phVoltage;
  }

  float phValue = constrain(7.0 + PH_SLOPE * (phVoltage - voltage_pH7), 0.0, 14.0);
  smoothPh = smooth(smoothPh, phValue);
  phValue = smoothPh;
  smoothingPrimed = true;
  bool phPlausible = (phValue >= PH_PLAUSIBLE_MIN && phValue <= PH_PLAUSIBLE_MAX);

  // ---------------- 5. Serial Monitor ----------------
  Serial.print("T=");
  if (tempOk) { Serial.print(temperature, 1); } else { Serial.print("25.0*"); }
  Serial.print(" C  Turb=");   Serial.print(turbidity, 1);
  Serial.print(" NTU  TDS=");  Serial.print(tdsValue, 0);
  Serial.print(" ppm  pH=");   Serial.print(phValue, 2);
  Serial.print("   [raw turb="); Serial.print(turbADC);
  Serial.print(" tds=");         Serial.print(tdsADC);
  Serial.print(" ph=");          Serial.print(phADC);
  Serial.print(" phV=");         Serial.print(phVoltage, 3);
  Serial.print("]");
  if (!phPlausible) Serial.print("  <- pH out of range, probe in the water?");
  Serial.println();

  lastTemp = temperature; lastTurb = turbidity; lastTds = tdsValue; lastPh = phValue;

  // ---------------- 6. Send one sample to the model ----------------
  // Axis order MUST match the axis order used when training in Edge Impulse:
  // ph, turbidity_ntu, tds_mgl, temp_c
  Bridge.notify("record_water_sample", phValue, turbidity, tdsValue, temperature);
}

// Called from Python once the model has scored a full window.
void showResult(String label, float confidence) {
  Serial.print(">>> Water quality: ");
  Serial.print(label);
  Serial.print("  score=");
  Serial.println(confidence, 3);

  unsigned long now = millis();
  bool isAnomaly = (label == "ANOMALY");
  bool sendNow = false;

  if (firstResult) {
    sendNow = true;                       // first verdict after boot
  } else if (isAnomaly && now - lastAlertMillis >= ALERT_INTERVAL_MS) {
    sendNow = true;
  } else if (now - lastReportMillis >= REPORT_INTERVAL_MS) {
    sendNow = true;
  }
  if (!sendNow) return;

  int risk = riskOutOf10(confidence);
  const char *level = risk >= 7 ? "HIGH" : risk >= 4 ? "MODERATE" : "LOW";

  String msg = risk >= 4 ? "Water Quality Alert\n" : "Water Quality Report\n";
  msg += "Source: "; msg += SOURCE_NAME; msg += "\n";
  msg += "Risk: "; msg += risk; msg += "/10 ("; msg += level; msg += ")\n";
  msg += "pH "; msg += String(lastPh, 2);
  msg += "  Turb "; msg += String(lastTurb, 1); msg += " NTU";
  msg += "  TDS "; msg += String(lastTds, 0); msg += " ppm";
  msg += "  T "; msg += String(lastTemp, 1); msg += " C\n";
  msg += "Time: "; msg += DUMMY_DATETIME;

  Serial.print(">>> Risk "); Serial.print(risk); Serial.print("/10 "); Serial.println(level);

  if (!sendSms(msg)) return;   // leave timers untouched so the next window retries
  firstResult = false;
  lastReportMillis = now;
  if (isAnomaly) lastAlertMillis = now;
}

profiles:
  default:
    fqbn: 
    platforms:
      - platform: arduino:zephyr
    libraries:
      - OneWire (2.3.8)
      - DallasTemperature (4.0.4)

default_profile: default

"""Water quality classification on the UNO Q.

The sketch samples the sensors and pushes one row (pH, turbidity, TDS,
temperature) per interval over the Router Bridge. We buffer rows until we have a full
model window, run inference on the Edge Impulse model served by the
motion_detection brick's ei-models-runner container, and send the winning label
back to the MCU so it shows up in the Serial Monitor.
"""

import os
import requests

from arduino.app_utils import App, Bridge, Logger

logger = Logger("WaterQualityAI")

# Axis order MUST match the column/axis order used when training in Edge Impulse.
AXES = ["ph", "turbidity_ntu", "tds_mgl", "temp_c"]

# Minimum confidence before we report a label to the MCU.
CONFIDENCE_THRESHOLD = float(os.getenv("CONFIDENCE_THRESHOLD", "0.6"))

# Anomaly-detection impulses have no labels; anything at or above this score is
# flagged. Use the same threshold you picked in Studio (0.3 is the EI default).
ANOMALY_THRESHOLD = float(os.getenv("ANOMALY_THRESHOLD", "0.3"))

# The runner started by the arduino:motion_detection brick.
# Inside the app container it is reachable by service name; on the host it is
# published on 127.0.0.1:1341.
RUNNER_URLS = [
    os.getenv("EI_RUNNER_URL", ""),
    "http://ei-motion-detection-runner:1337",
    "http://127.0.0.1:1341",
]


def _find_runner() -> str:
    for url in RUNNER_URLS:
        if not url:
            continue
        try:
            r = requests.get(f"{url}/api/info", timeout=5)
            if r.status_code == 200:
                return url
        except requests.RequestException:
            continue
    raise RuntimeError(
        "Edge Impulse runner not reachable. Is the motion_detection brick "
        "running and EI_MOTION_DETECTION_MODEL pointing at your .eim file?"
    )


RUNNER_URL = _find_runner()
INFO = requests.get(f"{RUNNER_URL}/api/info", timeout=10).json()
PARAMS = INFO["modelParameters"]

INPUT_FEATURES_COUNT = int(PARAMS["input_features_count"])
LABELS = PARAMS.get("labels", [])
FREQUENCY = PARAMS.get("frequency", -1)
SAMPLES_PER_WINDOW = max(1, INPUT_FEATURES_COUNT // len(AXES))

logger.info(f"model: {INFO.get('project', {}).get('name')}")
logger.info(
    f"labels={LABELS} input_features_count={INPUT_FEATURES_COUNT} "
    f"-> {SAMPLES_PER_WINDOW} sample(s) x {len(AXES)} axes, frequency={FREQUENCY} Hz"
)
logger.info("Set SAMPLE_INTERVAL_MS in the sketch to 1000/frequency ms.")

window: list[float] = []


def record_water_sample(ph: float, turbidity: float, tds: float, temperature: float):
    """Called by the sketch via Bridge.notify once per sample."""
    global window

    sample = [float(ph), float(turbidity), float(tds), float(temperature)]
    window.extend(sample)

    if len(window) < INPUT_FEATURES_COUNT:
        logger.info(
            f"buffering {len(window) // len(AXES)}/{SAMPLES_PER_WINDOW} samples"
        )
        return

    features = window[:INPUT_FEATURES_COUNT]
    window = window[INPUT_FEATURES_COUNT:]  # non-overlapping windows

    try:
        res = requests.post(
            f"{RUNNER_URL}/api/features", json={"features": features}, timeout=10
        ).json()
    except requests.RequestException as e:
        logger.error(f"inference request failed: {e}")
        return

    result = res.get("result", {})

    if "classification" in result:
        scores = result["classification"]
        logger.info("scores: " + "  ".join(f"{k}={v:.3f}" for k, v in scores.items()))
        label, confidence = max(scores.items(), key=lambda kv: kv[1])
        if confidence >= CONFIDENCE_THRESHOLD:
            Bridge.call("show_result", label, float(confidence))
        else:
            Bridge.call("show_result", "uncertain", float(confidence))

    if "regression" in result:
        value = float(result["regression"]["value"])
        logger.info(f"regression: {value:.3f}")
        Bridge.call("show_result", f"{value:.2f}", 1.0)

    if "anomaly" in result:
        anomaly = float(result["anomaly"])
        logger.info(f"anomaly score: {anomaly:.3f}")
        # No labels on an anomaly-detection impulse: report the score itself,
        # plus a verdict based on the threshold you picked in Studio.
        verdict = "ANOMALY" if anomaly >= ANOMALY_THRESHOLD else "normal"
        Bridge.call("show_result", verdict, anomaly)


Bridge.provide("record_water_sample", record_water_sample)

App.run()

requests