Published May 30, 2021

Wapr - a water profiler

Ever wonder which appliance is responsible for that humongous water bill? Me too. Introducing wapr: a non-invasive water usage-profiler.

IntermediateShowcase (no instructions)465

Things used in this project

Hardware components

QuickLogic Corp. QuickFeather Development Kit

Adafruit HUZZAH32 – ESP32 Feather Board

Software apps and online services

SensiML Analytics Toolkit

InfluxDB

Eclipse Mosquitto

Story

TL;DR

Identifying the active water consumers by measuring the vibrations on the main water pipe. Combining this information with the flow rate, we can create a water profile for each appliance and get insights on average water consumption, running time and potential leakages.

The Idea

It's a sad thing when you get your water bill at the end of the year and you notice you've used up enough water to fill an Olympic-sized swimming pool (which you don't have). This hurts in many ways: you have to pay extra for the water you unintentionally used, you feel bad for wasting the water and you feel frustrated because you only discover this now after weeks or even months...

Determined to never let this happen again you want to start monitoring your water usage at regular basis and keep an eye on which device uses how much water. Maybe it's time to replace that dishwasher with a newer model? But how to do this with a basic water meter?

Noting the water count each week with "pen & paper" will not survive the month (let's be honest). Installing a new digital water meter would be ideal, but the costs are high and you would have to contact the water company (or I don't know who) to install it, which for me is too much hassle because I'm just renting the place. And how to detect the different water appliances?

So I had this idea, what if we could make an IoT Edge device which analyses the vibrations of the water flowing through the main water pipe. With these vibrations we could identify the appliance which is using the water and do an estimation of the water consumption. This solution would require no modification of the existing installation and keep the budget and hassle as low as possible.

1. Objectives & Constraints

First things first, clear objectives and some constraints.

For this prototype I want to detect following things:

Water being used by the faucet
Water being used by the toilet (filling of the cistern)
Water being used by the shower
No water being used, idle

And generate reports:

Water consumption per device with an overview per day and month
On-duration per device with an overview per day and month
Alert notifications on abnormal water usage

The estimation of the water consumption of each device will be done by multiplying the running time with a fixed measured flow rate (fill a bucket, see how long it takes to get a liter). Can't we detect this as well you ask? Hmm yes maybe, but let's first learn how to walk before we start to run.

Following that same mindset, I've given myself some slack by adding some constraints:

Only one water consumer is active at the same time
Central heating off: I've got an ancient boiler which generates a lot of vibrations and is right next to my central water line.

2. Training - Hardware Setup

Just a couple of hardware components are used:

QuickLogic Corp's QuickFeather: The brains and power behind the project, it contains the accelerometers and will run the machine learning model on board to do the predictions. Special thanks to QuickLogic Corp for supplying the hardware!
Adafruit HUZZAH32 - ESP32 Feather Board for adding some connectivity over WiFi
A 3D-printed case for mounting the boards in, files provided by adafruit right here!

Once assembled, attach it to the main water line inside your house/apartment/cabin/mansion/... You'll also need to supply power using the micro-USB on the feather, so make sure you have a power socket nearby.

I've started with some cable ties and electrical tape to attach it to the water pipe, but eventually I added a part which snaps on the water line and has velcro tape for an extra rigid connection (stl-file can be found in the attachments).

Casing with velcro strap and snap-on module - Installation on the water line

Tip: Before applying a bottle of Gorilla super glue for a permanent installation, skip ahead to the Software Setup - Training section in this document and verify if everything works correctly before installing it at location.

Tip: Don't use Gorilla (or any other) super glue, I was joking.

3. Training - Software

For collecting data, labeling and training models, I've used the tools provided by SensiML which make the whole process easy and effortless, no need to be a machine learning expert to make this work. They have great documentation which you can find here & here & here & here.

3. Deployment - Software

wapr architecture

Once the model was trained, I uploaded the binary file to the QuickFeather and uploaded a custom program to the ESP32 to post the classification output to a MQTT server. A MQTT-Client application in python listens for these classifications and uploads them to an influxdb instance. Influxdb 2.0 comes with a complete platform for data exploration and monitoring, worth while to check it out: https://www.influxdata.com/

Besides the data-ingestion application, I also added a data-calculator application which takes data from influxdb, does some simple math stuff - such as calculating the water usage, total running time,... - and uploads it to influxdb as new records. I'm pretty sure you could some of these tasks also in influxdb with some flux-queries, but for me it was easier to write some code for it.

I also added the possibility to do a soft reset of the QuickFeather by publishing a message to a reset-MQTT-topic to which the ESP32 is subscribed to. This is needed because after a while - 15 minutes or so - the QuickFeather started to output jibberish/unknown classification outputs. When the ingestion script receives messages like that, it will publish a reset message so the QuickFeather would reset and reinitialize the classification. This is more a work around than really solving the issue, but it works for me.

If you're interested you can take a look at the code in the attachment, I've included the python scripts and the source file for the feather.

4. Deployment - Hardware Adjustments

For resetting the QuickFeather without resetting the ESP32, I had to disconnect the connection between the reset pins, which were connected through the header pins. I pushed the header pin of the QuickFeather up and connected it with an output of the ESP32.

5. Reporting

In order to evaluate and use the data coming from the sensor, I created a couple of dashboards in influxdb:

A model-dashboard containing the raw output of the model and some derived graphs for making it easier to understand

wapr model-dashboard

An overview-dashboard containing the amount of water used and the on-duration per device; near real time, per day and per month.

wapr overview-dashboard with from left to right the water consumed in liters that day by the faucet, shower and toilet.

This water usage is an estimate based on the on-time detected by the model and the flow-rate which I determined with a bucket and a stopwatch. As a check I noted the values of the water-meter before and after I flushed the toilet which confirmed my estimate:

Water consumption estimate of toilet flushing and actual water-meter values.

6. Evaluation of the objectives

This is what it all boils down to, does it even work? Here are my honest findings:

Water being used by the toilet (filling of the cistern)

toilet classification - 0 toilet not running - 1 toilet running

Works like a charm: no noticeable delay in the detection, no false positives or negatives. I've simulated a running toilet by displacing the sealing ring of the cistern and the model picked it up without a problem.

Water being used by the faucet

faucet classification - 0 faucet not running - 1 faucet running

Overall good results: False negatives - faucet is running but not detected - barely exist, and when they occur it quickly recovers and detects the running faucet. False positives - faucet is not running but the classifier says otherwise - only happens when the shower is running. More about this in the shower objective evaluation.

Water being used by the shower

Ah the problem child... When the shower is running the model gets confused and can't make up it's mind if it's the shower, idle or the faucet that is currently running. So in the plot we'll see the output vary a lot, no bueno.

A tiny silver lining is though that its obvious when the shower is being used by looking at the graphs. I played with the idea of using this data to train an additional ml-model which would - when given a window of predictions - a classification of 'shower-on' or 'shower-off'. But I decided against it because it's not really a solution, it's more a huge dirty flimsy workaround. The only proper way to tackle this issue is more and better data and a lot more testing, which is somewhere on the to do list.

No water being used, idle

idle classification - 0 not idle - 1 idle. This plot contains a faucet event where the idle classification is thus 0.

I want to say that this works perfectly, but I can't due to the issue we discussed with the shower. But besides that, pretty darn good!

7. Evaluation of the constraints

In the beginning we've added some constraints to the project in order to make the problem a bit easier. Now what would happen if we would forget about these constraints and use the model as is? Is it still valid?

Only one water consumer is active at the same time

I would say for the most of the time, the model would still be correct in my case. I live alone, so small chances that I'll be flushing the toilet and be in the shower at the same time.

Central heating off: I've got an ancient boiler which generates a lot of vibrations and is right next to my central water line.

This would mess with the model. Apparently the vibrations of the boiler heating the water seem like a running faucet accordingly to the model. It is only normal that the model doesn't know what to do with this because I didn't add training data of this scenario.

I did notice something interesting while running the test with the boiler on, apparently it keeps the internal hot water coil on a constant temperature -the comfort mode according to the manual - for which it turns on every 10 minutes and heats the water.

Classification with boiler on between 1 and 6 am - Periodically heating each 10 min for 2 min

This is totally not needed for me because I only use hot water twice a day (shower & dishes), so now I turn this bad boy on manually if I need it and by doing so saving on my gas bill.

8. Conclusion

The main objective was to determine the active water users solely by using the vibrations of the main waterline, and I feel I can say this is accomplished if you give me some slack concerning the problems with the shower.

This project surely proves that it is feasible to create such an edge device, but this will not be a plug-and-play installation because each water installation is different and thus the ml-model must be retrained for each case. Maybe someday wapr will exist with a fancy app to retrain it in a user-friendly way, but until then you'll have to go through this blog!

Peace!

9. Notes: Improvements for wapr2.0

Some quick notes for future me, suggestions are welcome:

Experiment with a higher sampling frequency (currently 100Hz which seem a bit on the low side)
Use generated ml-model output to automatically label new data (semi-supervised learning?)
Upgrade hardware connection to the water line, experiment with different attachments

Code

#include <Arduino.h>
#include <Wire.h>
#include "WaprMQTT.h"
#include <ArduinoJson.h>

// Hardware ESP32
#ifdef ARDUINO_ARCH_ESP32
#define RXD2 17
#define TXD2 16
#define RESET_FEATHER_PIN 13
#endif

// Classification
uint8_t classification[1];

// Deserialization
String jsonString;
StaticJsonDocument<768> doc;

// Feather reset
int resetDuration = 1000;
int resetDelay = 5000;

// Communication - MQTT
WaprMQTT comClient;

void resetFeather(){
  Serial.println("resetting feather..");
  digitalWrite(RESET_FEATHER_PIN, LOW);
  delay(resetDuration);
  digitalWrite(RESET_FEATHER_PIN, HIGH);
  delay(resetDelay);
}

void setup() {
  Serial.begin(9600);
  Serial.println("running setup..");
  Serial.println("initializing pins..");
  pinMode(RESET_FEATHER_PIN, OUTPUT);
  digitalWrite(RESET_FEATHER_PIN, HIGH);

  Serial.println("intilializing communication..");
  comClient.init();
  
  Serial.println("setting up serial 2..");
  Serial2.begin(460800, SERIAL_8N1, RXD2, TXD2);

  Serial.println("entering loop..");
  delay(100);
}

void loop() {
  if(Serial2.available()>0){
    jsonString = Serial2.readStringUntil('\n');
    DeserializationError error = deserializeJson(doc, jsonString);

    if (error) {
      Serial.print(F("deserializeJson() failed: "));
      Serial.println(error.f_str());

      comClient.update(DEBUG, classification, sizeof(classification));
    }
    else{
      classification[0] = doc["Classification"];
      Serial.println(classification[0]);
      comClient.update(CLASSIFICATION, classification, sizeof(classification));
      if(int(classification[0])==0){
        resetFeather();
      }
    }
  }

  comClient.loop();
}

/*
    Interface for communicating.
*/
#ifndef _WaprMQTT_H
#define _WaprMQTT_H
#include <Arduino.h>

#define MQTT_TOPIC_CLASSIFICATION "wapr/proto/classification"
#define MQTT_TOPIC_DEBUG "wapr/proto/debug"
#define MQTT_TOPIC_RESTART "wapr/proto/restart"
#define MQTT_TOPIC_TEST "wapr/test"

typedef enum {
    CLASSIFICATION,
    TEST,
    DEBUG
} TOPIC;

class WaprMQTT {
public:
    WaprMQTT();
    virtual void init();
    virtual bool update(TOPIC topic, uint8_t bytes[], uint8_t size);
    void loop();
private:
    void setupWifi();
    void reconnect();
    static void callback(char* topic, byte* message, unsigned int length);
};

#endif

#include "WaprMQTT.h"
#include <Arduino.h>
#include <WiFi.h>
#include <PubSubClient.h>

const char* ssid = "SSID";
const char* password = "WIFI_PASSWORD";
const char* mqtt_server = "IP_MQTT_SERVER";

char classificationPayload[1];
char testPayload[1];
char debugPayload[1];

WiFiClient wifiClient;
PubSubClient mqttClient(wifiClient);

WaprMQTT::WaprMQTT(){

}


void WaprMQTT::callback(char* topic, byte* message, unsigned int length) {
  Serial.print("message arrived on topic: ");
  Serial.print(topic);
  Serial.print(". message: ");
  String messageTemp;
  
  for (int i = 0; i < length; i++) {
    Serial.print((char)message[i]);
    messageTemp += (char)message[i];
  }
  Serial.println();

  if(String(topic) == MQTT_TOPIC_RESTART){
    Serial.println("rebooting quickfeather..");
    ESP.restart();
  }

  if (String(topic) == "esp32/output") {
    Serial.print("Changing output to ");
    if(messageTemp == "on"){
      Serial.println("on");
      // digitalWrite(ledPin, HIGH);
    }
    else if(messageTemp == "off"){
      Serial.println("off");
      // digitalWrite(ledPin, LOW);
    }
  }

  
}

void WaprMQTT::reconnect() {
  while (!mqttClient.connected()) {
    Serial.println("Attempting MQTT connection...");
    if (mqttClient.connect("WaprProto")) {
      Serial.println("connected");
      // ... and resubscribe
      mqttClient.subscribe("wapr/proto/debug");
      mqttClient.subscribe(MQTT_TOPIC_RESTART);

    } 
    else {
      Serial.print("failed, rc=");
      Serial.print(mqttClient.state());
      Serial.println(" trying again in 5 seconds..");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void WaprMQTT::loop(){

    if (!mqttClient.connected())
    {
        reconnect();
    }

    mqttClient.loop();
}

void WaprMQTT::setupWifi(){
    delay(10);
    // We start by connecting to a WiFi network
    Serial.print("connecting to ");
    Serial.print(ssid);
    Serial.println("..");
    WiFi.begin(ssid, password);

    while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
    }

    Serial.println("");
    Serial.print("connected with IP address: ");
    Serial.println(WiFi.localIP());
}


void WaprMQTT::init() {
    setupWifi();
    mqttClient.setServer(mqtt_server, 1883);
    mqttClient.setCallback(callback);
    mqttClient.subscribe("wapr/proto/debug");
    mqttClient.subscribe(MQTT_TOPIC_RESTART);
}

bool WaprMQTT::update(TOPIC topic, uint8_t bytes[], uint8_t size) {
    char* pPayload;
    switch(topic){
        case CLASSIFICATION:
            pPayload = &classificationPayload[0];
            for (int i = 0; i < size; i++)
            {
                sprintf(pPayload, "%02X", bytes[i]);
                pPayload+=2;
            }
            mqttClient.publish(MQTT_TOPIC_CLASSIFICATION, classificationPayload);
            return true;
        case TEST:
            pPayload = &testPayload[0];
            for (int i = 0; i < size; i++)
            {
                sprintf(pPayload, "%02X", bytes[i]);
                pPayload+=2;
            }
            mqttClient.publish(MQTT_TOPIC_TEST, testPayload);
            return true;
        case DEBUG:
            pPayload = &debugPayload[0];
            for (int i = 0; i < size; i++)
            {
                sprintf(pPayload, "%02X", bytes[i]);
                pPayload+=2;
            }
            mqttClient.publish(MQTT_TOPIC_TEST, debugPayload);
            return true;
        default:
            return false;
    }
}

from datetime import datetime
import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import time

CLASSIFICATION_UNKNOWN = 0
CLASSIFICATION_FAUCET = 1
CLASSIFICATION_IDLE = 2
CLASSIFICATION_SHOWER = 3
CLASSIFICATION_TOILET = 4

classificationFailCount = 0

def on_message(client, userdata, message):
    if(message.topic == classificationTopic):
        global classificationFailCount
        result = int(message.payload.decode("utf-8"))
        p = []
        p.append(Point("quickfeater").field("classification", result))
        if(result == CLASSIFICATION_IDLE):
            classificationFailCount = 0
            p.append(Point("quickfeater").field("idle", 1))
            p.append(Point("quickfeater").field("faucet", 0))
            p.append(Point("quickfeater").field("shower", 0))
            p.append(Point("quickfeater").field("toilet", 0))
        elif(result == CLASSIFICATION_FAUCET):
            classificationFailCount = 0
            p.append(Point("quickfeater").field("idle", 0))
            p.append(Point("quickfeater").field("faucet", 1))
            p.append(Point("quickfeater").field("shower", 0))
            p.append(Point("quickfeater").field("toilet", 0))
        elif(result == CLASSIFICATION_SHOWER):
            classificationFailCount = 0
            p.append(Point("quickfeater").field("idle", 0))
            p.append(Point("quickfeater").field("faucet", 0))
            p.append(Point("quickfeater").field("shower", 1))
            p.append(Point("quickfeater").field("toilet", 0))
        elif(result == CLASSIFICATION_TOILET):
            classificationFailCount = 0
            p.append(Point("quickfeater").field("idle", 0))
            p.append(Point("quickfeater").field("faucet", 0))
            p.append(Point("quickfeater").field("shower", 0))
            p.append(Point("quickfeater").field("toilet", 1))
        elif(result == CLASSIFICATION_UNKNOWN):
            print(f'unknown classification: {result}')
            classificationFailCount += 1
            if(classificationFailCount>10):
                print('something is wrong, rebooting quickfeather..')
                client.publish(restartTopic, "1")
                time.sleep(10)
        write_api.write(bucket=bucket, record=p)
    else:
        print(f'unknown mqtt topic {message.topic} sends {message.payload.decode("utf-8")}')

print("influx ingestion app starting..")
print("connecting to influxdb..")
bucket="YOUR_BUCKET"
token = "YOUR_INFLUXDB_TOKEN"
org = "YOUR_ORG"
client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)

# topics
testTopic = "wapr/test"
classificationTopic = "wapr/proto/classification"
restartTopic = "wapr/proto/restart"

# setup client connection
print("connecting to mqtt..")
client = mqtt.Client("influxIngestWapr")
client.connect("MQTT_IP")
print("subscribing to topics..")
client.subscribe(classificationTopic)
client.subscribe(testTopic)
client.on_message = on_message
print("starting client loop..")
client.loop_start()

input("press any key to stop..")

client.loop_stop()

print("goodbye")

from datetime import datetime, timezone
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import time
import pytz
import threading


# flow rates in liter/second
FLOW_RATE_FAUCET = 0.1186
FLOW_RATE_SHOWER = 0.1345
FLOW_RATE_TOILET = 0.1049

faucetWindow = 5
idleWindow = 5
toiletWindow = 5
showerWindow = 5

wFaucetThreshold = 0.4
wToiletThreshold = 0.3
wShowerTheshold = 0.3

faucetThreshold = 0.0
toiletThreshold = 0.0
showerTheshold = 0.0

wFaucetPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 
wToiletPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 
wShowerPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 

faucetPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 
toiletPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 
showerPreviousQueryTime = datetime(1970, 1, 1, tzinfo=timezone.utc) 


timeRangeStartSearchPrevious = "-6h"
timeRangeStart = "-5m"
timeRangeStop = "-10s"
secondOfSleep = 250

keepRunning = True

#region Queries
queryFaucetWindow = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "faucet")\
  |> aggregateWindow(every: {faucetWindow}s, fn: mean, createEmpty: false)\
  |> yield(name: "mean")'

queryFaucet = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "faucet")'

queryFaucetWindowPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyFaucetOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")'

queryFaucetPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyFaucetOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryFaucetWindowConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyFaucetWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")'

queryFaucetConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyFaucetWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryToiletWindow = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "toilet")\
  |> aggregateWindow(every: {toiletWindow}s, fn: mean, createEmpty: false)\
  |> yield(name: "mean")'

queryToilet = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "toilet")'

queryToiletWindowPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyToiletOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")'

queryToiletPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyToiletOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryToiletWindowConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyToiletWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")\
  |> aggregateWindow(every: 5s, fn: last, createEmpty: false)\
  |> yield(name: "last")'

queryToiletConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyToiletWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryShowerWindow = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "shower")\
  |> aggregateWindow(every: {showerWindow}s, fn: mean, createEmpty: false)\
  |> yield(name: "mean")'

queryShower = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "shower")'

queryShowerWindowPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyShowerOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")'

queryShowerPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyShowerOnDurationFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryShowerWindowConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyShowerWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "True")'

queryShowerConsumptionPreviousTime = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStartSearchPrevious}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "waprCalculator")\
  |> filter(fn: (r) => r["_field"] == "dailyShowerWaterConsumptionFloat")\
  |> filter(fn: (r) => r["windowed"] == "False")'

queryIdleWindow = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "idle")\
  |> aggregateWindow(every: {idleWindow}s, fn: mean, createEmpty: false)\
  |> yield(name: "mean")'

queryIdle = f'from(bucket: "wapr")\
  |> range(start: {timeRangeStart}, stop: {timeRangeStop})\
  |> filter(fn: (r) => r["_measurement"] == "quickfeater")\
  |> filter(fn: (r) => r["_field"] == "idle")'
#endregion

def getPreviousQueryTime(result):
    '''
    Get the latest time value of the result
    '''
    for table in result:
        if(len(table.records)>0):
            lastRecord = table.records[-1]
            time = lastRecord.get_time()
            return time
    return datetime(1970, 1, 1, tzinfo=timezone.utc)

def getLastestValue(result):
    '''
    Get the latest value of the result
    '''
    if(len(result)==0):
      return 0
    for table in result:
      if(len(table.records)>0):
        lastRecord = table.records[-1]
        value = lastRecord.get_value()
        return value
      else:
        print('something is wrong, value not found')
        return -1
  

def calcWaterUsage(durationOn, flowRate):
    '''
    Returns the amount of liters used for the duration in seconds with flow rate in liters per second
    '''
    return (durationOn * flowRate)

def calcDurationOn(result, threshold, lastQueryTime):
    '''
    Calculates the total duration in seconds that the value was higher than the threshold
    '''
    on = False
    durationOn = 0 # in seconds
    startTime = datetime.now()
    lastOnTime = datetime.now()
    for table in result:
        for record in table.records:
            if(record.get_time()>lastQueryTime):
                if(record.get_value()>threshold):
                    if(on == False):
                        on = True
                        startTime = record.get_time()
                        lastOnTime = startTime
                    else:
                        lastOnTime = record.get_time()
                else:
                    if(on == True):
                        on = False
                        duration = (lastOnTime - startTime).total_seconds()
                        durationOn += duration
    if(on == True):
        on = False
        duration = (lastOnTime - startTime).total_seconds()
        durationOn += duration
    return durationOn
        
def runCalculation():
    '''
    Loop through this function
    '''
    print('')
    print('running calculation...')
    global keepRunning
    global wFaucetPreviousQueryTime
    global wToiletPreviousQueryTime
    global wShowerPreviousQueryTime
    global faucetPreviousQueryTime
    global toiletPreviousQueryTime
    global showerPreviousQueryTime

    # first time get the latest values
    wLatestFaucetDurationCalculation = client.query_api().query(org=org, query=queryFaucetWindowPreviousTime)
    wLatestToiletDurationCalculation = client.query_api().query(org=org, query=queryToiletWindowPreviousTime)
    wLatestShowerDurationCalculation = client.query_api().query(org=org, query=queryShowerWindowPreviousTime)
    latestFaucetDurationCalculation = client.query_api().query(org=org, query=queryFaucetPreviousTime)
    latestToiletDurationCalculation = client.query_api().query(org=org, query=queryToiletPreviousTime)
    latestShowerDurationCalculation = client.query_api().query(org=org, query=queryShowerPreviousTime)

    currentUTC = datetime.now(timezone.utc)
    print(f'current utc: {currentUTC}')
    wFaucetPreviousQueryTime = getPreviousQueryTime(wLatestFaucetDurationCalculation)
    wToiletPreviousQueryTime = getPreviousQueryTime(wLatestToiletDurationCalculation)
    wShowerPreviousQueryTime = getPreviousQueryTime(wLatestShowerDurationCalculation)
    print(f'wFaucetPreviousQueryTime:{wFaucetPreviousQueryTime}')
    print(f'wToiletPreviousQueryTime:{wToiletPreviousQueryTime}')
    print(f'wShowerPreviousQueryTime:{wShowerPreviousQueryTime}')
    faucetPreviousQueryTime = getPreviousQueryTime(latestFaucetDurationCalculation)
    toiletPreviousQueryTime = getPreviousQueryTime(latestToiletDurationCalculation)
    showerPreviousQueryTime = getPreviousQueryTime(latestShowerDurationCalculation)
    print(f'faucetPreviousQueryTime:{faucetPreviousQueryTime}')
    print(f'toiletPreviousQueryTime:{toiletPreviousQueryTime}')
    print(f'showerPreviousQueryTime:{showerPreviousQueryTime}')

    wFaucetLatestOnDurationValue = getLastestValue(wLatestFaucetDurationCalculation)
    wToiletLatestOnDurationValue = getLastestValue(wLatestToiletDurationCalculation)
    wShowerLatestOnDurationValue = getLastestValue(wLatestShowerDurationCalculation)
    print(f'wFaucetLatestOnDurationValue:{wFaucetLatestOnDurationValue} seconds')
    print(f'wToiletLatestOnDurationValue:{wToiletLatestOnDurationValue} seconds')
    print(f'wShowerLatestOnDurationValue:{wShowerLatestOnDurationValue} seconds')
    faucetLatestOnDurationValue = getLastestValue(latestFaucetDurationCalculation)
    toiletLatestOnDurationValue = getLastestValue(latestToiletDurationCalculation)
    showerLatestOnDurationValue = getLastestValue(latestShowerDurationCalculation)
    print(f'faucetLatestOnDurationValue:{faucetLatestOnDurationValue} seconds')
    print(f'toiletLatestOnDurationValue:{toiletLatestOnDurationValue} seconds')
    print(f'showerLatestOnDurationValue:{showerLatestOnDurationValue} seconds')

    wLatestFaucetConsumptionCalculation = client.query_api().query(org=org, query=queryFaucetWindowConsumptionPreviousTime)
    wLatestToiletConsumptionCalculation = client.query_api().query(org=org, query=queryToiletWindowConsumptionPreviousTime)
    wLatestShowerConsumptionCalculation = client.query_api().query(org=org, query=queryShowerWindowConsumptionPreviousTime)
    latestFaucetConsumptionCalculation = client.query_api().query(org=org, query=queryFaucetConsumptionPreviousTime)
    latestToiletConsumptionCalculation = client.query_api().query(org=org, query=queryToiletConsumptionPreviousTime)
    latestShowerConsumptionCalculation = client.query_api().query(org=org, query=queryShowerConsumptionPreviousTime)

    wFaucetLatestConsumptionValue = getLastestValue(wLatestFaucetConsumptionCalculation)
    wToiletLatestConsumptionValue = getLastestValue(wLatestToiletConsumptionCalculation)
    wShowerLatestConsumptionValue = getLastestValue(wLatestShowerConsumptionCalculation)
    print(f'wFaucetLatestConsumptionValue:{wFaucetLatestConsumptionValue} liters')
    print(f'wToiletLatestConsumptionValue:{wToiletLatestConsumptionValue} liters')
    print(f'wShowerLatestConsumptionValue:{wShowerLatestConsumptionValue} liters')
    faucetLatestConsumptionValue = getLastestValue(latestFaucetConsumptionCalculation)
    toiletLatestConsumptionValue = getLastestValue(latestToiletConsumptionCalculation)
    showerLatestConsumptionValue = getLastestValue(latestShowerConsumptionCalculation)
    print(f'faucetLatestConsumptionValue:{faucetLatestConsumptionValue} liters')
    print(f'toiletLatestConsumptionValue:{toiletLatestConsumptionValue} liters')
    print(f'showerLatestConsumptionValue:{showerLatestConsumptionValue} liters')

    currentDate = datetime.now()
    myTimezone = pytz.timezone("Europe/Brussels")
    currentDate = myTimezone.localize(currentDate)

    while(keepRunning == True):
        # check if it is a new day
        newDate = datetime.now()
        newDate = myTimezone.localize(newDate)

        if(newDate.date() != currentDate.date()):
          wFaucetLatestOnDurationValue = 0
          wToiletLatestOnDurationValue = 0
          wShowerLatestOnDurationValue = 0
          wFaucetLatestConsumptionValue = 0
          wToiletLatestConsumptionValue = 0
          wShowerLatestConsumptionValue = 0
          faucetLatestOnDurationValue = 0
          toiletLatestOnDurationValue = 0
          showerLatestOnDurationValue = 0
          faucetLatestConsumptionValue = 0
          toiletLatestConsumptionValue = 0
          showerLatestConsumptionValue = 0
          currentDate = newDate
        
        resultFaucetWindow = client.query_api().query(org=org, query=queryFaucetWindow)
        resultToiletWindow = client.query_api().query(org=org, query=queryToiletWindow)
        resultShowerWindow = client.query_api().query(org=org, query=queryShowerWindow)
        resultFaucet = client.query_api().query(org=org, query=queryFaucet)
        resultToilet = client.query_api().query(org=org, query=queryToilet)
        resultShower = client.query_api().query(org=org, query=queryShower)

        wFaucetOnDuration = calcDurationOn(resultFaucetWindow, wFaucetThreshold, wFaucetPreviousQueryTime)
        wToiletOnDuration = calcDurationOn(resultToiletWindow, wToiletThreshold, wToiletPreviousQueryTime)
        wShowerOnDuration = calcDurationOn(resultShowerWindow, wShowerTheshold, wShowerPreviousQueryTime)
        print(f'windowed faucetOnDuration:{wFaucetOnDuration}')
        print(f'windowed toiletOnDuration:{wToiletOnDuration}')
        print(f'windowed showerOnDuration:{wShowerOnDuration}')
        faucetOnDuration = calcDurationOn(resultFaucet, faucetThreshold, faucetPreviousQueryTime)
        toiletOnDuration = calcDurationOn(resultToilet, toiletThreshold, toiletPreviousQueryTime)
        showerOnDuration = calcDurationOn(resultShower, showerTheshold, showerPreviousQueryTime)
        print(f'faucetOnDuration:{faucetOnDuration}')
        print(f'toiletOnDuration:{toiletOnDuration}')
        print(f'showerOnDuration:{showerOnDuration}')

        now = datetime.now(timezone.utc)
        wFaucetPreviousQueryTime = now
        wToiletPreviousQueryTime = now
        wShowerPreviousQueryTime = now
        wFaucetLatestOnDurationValue += wFaucetOnDuration
        wToiletLatestOnDurationValue += wToiletOnDuration
        wShowerLatestOnDurationValue += wShowerOnDuration
        faucetPreviousQueryTime = now
        toiletPreviousQueryTime = now
        showerPreviousQueryTime = now
        faucetLatestOnDurationValue += faucetOnDuration
        toiletLatestOnDurationValue += toiletOnDuration
        showerLatestOnDurationValue += showerOnDuration
        
        wFaucetWaterConsumption = calcWaterUsage(wFaucetOnDuration, FLOW_RATE_FAUCET)
        wToiletWaterConsumption = calcWaterUsage(wToiletOnDuration, FLOW_RATE_TOILET)
        wShowerWaterConsumption = calcWaterUsage(wShowerOnDuration, FLOW_RATE_SHOWER)
        print(f'windowed faucetWaterConsumption:{wFaucetWaterConsumption} liters')
        print(f'windowed toiletWaterConsumption:{wToiletWaterConsumption} liters')
        print(f'windowed showerWaterConsumption:{wShowerWaterConsumption} liters')
        faucetWaterConsumption = calcWaterUsage(faucetOnDuration, FLOW_RATE_FAUCET)
        toiletWaterConsumption = calcWaterUsage(toiletOnDuration, FLOW_RATE_TOILET)
        showerWaterConsumption = calcWaterUsage(showerOnDuration, FLOW_RATE_SHOWER)
        print(f'faucetWaterConsumption:{faucetWaterConsumption} liters')
        print(f'toiletWaterConsumption:{toiletWaterConsumption} liters')
        print(f'showerWaterConsumption:{showerWaterConsumption} liters')

        wFaucetLatestConsumptionValue += wFaucetWaterConsumption
        wToiletLatestConsumptionValue += wToiletWaterConsumption
        wShowerLatestConsumptionValue += wShowerWaterConsumption
        faucetLatestConsumptionValue += faucetWaterConsumption
        toiletLatestConsumptionValue += toiletWaterConsumption
        showerLatestConsumptionValue += showerWaterConsumption

        p = []
        p.append(Point("waprCalculator").field("dailyFaucetOnDurationFloat", float(wFaucetLatestOnDurationValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyToiletOnDurationFloat", float(wToiletLatestOnDurationValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyShowerOnDurationFloat", float(wShowerLatestOnDurationValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyFaucetWaterConsumptionFloat", float(wFaucetLatestConsumptionValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyToiletWaterConsumptionFloat", float(wToiletLatestConsumptionValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyShowerWaterConsumptionFloat", float(wShowerLatestConsumptionValue)).tag("windowed",True))
        p.append(Point("waprCalculator").field("dailyFaucetOnDurationFloat", float(faucetLatestOnDurationValue)).tag("windowed",False))
        p.append(Point("waprCalculator").field("dailyToiletOnDurationFloat", float(toiletLatestOnDurationValue)).tag("windowed",False))
        p.append(Point("waprCalculator").field("dailyShowerOnDurationFloat", float(showerLatestOnDurationValue)).tag("windowed",False))
        p.append(Point("waprCalculator").field("dailyFaucetWaterConsumptionFloat", float(faucetLatestConsumptionValue)).tag("windowed",False))
        p.append(Point("waprCalculator").field("dailyToiletWaterConsumptionFloat", float(toiletLatestConsumptionValue)).tag("windowed",False))
        p.append(Point("waprCalculator").field("dailyShowerWaterConsumptionFloat", float(showerLatestConsumptionValue)).tag("windowed",False))
        write_api.write(bucket=bucket, record=p)

        time.sleep(secondOfSleep)


# app initialization 
print('starting wapr-calculator..')
print("connecting to influxdb..")
bucket="YOUR_BUCKET"
token = "YOUR_TOKEN"
org = "YOUR_ORG"
client = InfluxDBClient(url="http://localhost:8086", token=token, org=org)
write_api = client.write_api(write_options=SYNCHRONOUS)
query_api = client.query_api()

calcThread = threading.Thread(target=runCalculation)
calcThread.start()

input("press any key to stop the calculation thread")
keepRunning = False

input("press any key to exit")

Credits

bena

1 project • 1 follower

Wapr - a water profiler

Things used in this project

Hardware components

Software apps and online services

Story

TL;DR

The Idea

1. Objectives & Constraints

2. Training - Hardware Setup

3. Training - Software

3. Deployment - Software

4. Deployment - Hardware Adjustments

5. Reporting

6. Evaluation of the objectives

7. Evaluation of the constraints

8. Conclusion

9. Notes: Improvements for wapr2.0

Custom parts and enclosures

Add-on module feather case

Code

Feather main.cpp

Feather WaprMQTT.h

Feather WaprMQTT.cpp

Influx Ingestion Script

Wapr Calculator

Credits

bena

Comments

Embed the widget on your own site

Wapr - a water profiler

Wapr - a water profiler

Things used in this project

Hardware components

Software apps and online services

Story

TL;DR

The Idea

1. Objectives & Constraints

2. Training - Hardware Setup

3. Training - Software

3. Deployment - Software

4. Deployment - Hardware Adjustments

5. Reporting

6. Evaluation of the objectives

7. Evaluation of the constraints

8. Conclusion

9. Notes: Improvements for wapr2.0

Custom parts and enclosures

Add-on module feather case

Code

Feather main.cpp

Feather WaprMQTT.h

Feather WaprMQTT.cpp

Influx Ingestion Script

Wapr Calculator

Credits

bena

Comments

Related channels and tags