Защита от протечек с блекджеком и счетчиками

// This #include statement was automatically added by the Particle IDE.
#include "Adafruit_SSD1306/Adafruit_SSD1306.h"

// This #include statement was automatically added by the Particle IDE.
#include "MQTT/MQTT.h"

// This #include statement was automatically added by the Particle IDE.
#include "OneWire/OneWire.h"

SYSTEM_THREAD(ENABLED);
SYSTEM_MODE(MANUAL);
STARTUP(WiFi.selectAntenna(ANT_EXTERNAL));
STARTUP(System.enableFeature(FEATURE_RETAINED_MEMORY));

struct counter_struct {
  float value;
  byte state;
  int pin;
};
struct valve_struct {
  byte state;
  int pin;
};
struct sensor_struct {
  int timeout;
  byte state;
  int pin;
};

unsigned long currentMillis = 0;
unsigned long previous_conected = 100000; //финт ушами 
unsigned long previous_wifi_uptime = 100000; //финт ушами 
unsigned long previous_counter_read = 0; //финт ушами 
unsigned long wifi_uptime;
unsigned long start_temp_timer = 0;
unsigned long read_temp_timer = 0;
byte display_timeout = 0;

//temp onewire 
OneWire ds0 = OneWire(D2);
OneWire ds1 = OneWire(D3);
byte addr0[8];
byte addr1[8];
bool presense0 = false;
bool presense1 = false;
byte data[12];

#define OLED_RESET A7
Adafruit_SSD1306 display(OLED_RESET);


//valve control
retained valve_struct valve[2] = { {0, D4}, {0, D5} };

//counter control
retained counter_struct counter[2] = { {0, 1, A0}, {0, 1, A1} };
volatile int pressure[2] = {A2, A3};
#define SENSOR_TIMEOUT 10
volatile sensor_struct sensor[2] = { {0, 1, D6}, {0, 1, D7} };

void callback(char* topic, byte* payload, unsigned int length);

byte server[] = { 192,168,2,101};
MQTT client(server, 1883, callback);

bool publish_message(const char* t, const char* p, bool retain) 
{
    return client.publish(t, (uint8_t*)p, sizeof(p), retain);
}

bool publish_message(const char* t, int p, bool retain) 
{   
    char buf_d[12];
    int n = sprintf(buf_d,"%d",p);
    return client.publish(t, (uint8_t*)buf_d, n, retain);
}

bool publish_message(const char* t, float p, bool retain) 
{   
    //char buf_f[18];
    String s(p, 4);
//    dtostrf(p, 9, 4, buf_f);
    //int n = sprintf(buf_f,"%f",p);
    return client.publish(t, (uint8_t*)s.c_str(), s.length(), retain);
}

// recieve message
void callback(char* topic, byte* payload, unsigned int length) {
    char p[length + 1];
    memcpy(p, payload, length);
    p[length] = NULL;
    String message(p);
    String t(topic);
    if (t.equals("home/water_count/spark/set"))
    {
        if (message.equalsIgnoreCase("1"))
        {
            Particle.connect();
            if (waitFor(Particle.connected, 10000)) 
                {publish_message("home/water_count/spark", 1, false);}
            else
                {Particle.disconnect(); publish_message("home/water_count/spark", 0, false);}
        }
        else
        {
            Particle.disconnect();
            publish_message("home/water_count/spark", 0, false);
        }
    }
    else if (t.startsWith("home/water_count/valve/"))
    {
        int m = message.toInt();
        int x = t.substring(23,24).toInt();
        if (m > -1 && m < 2 && x > -1 && x <2) 
        {
            set_valve(x, m);
        }
        else
        {
            publish_message("home/water_count/valve/" + t.substring(23,24),  valve[x].state , true);
        }
    }
    else if (t.startsWith("home/water_count/counter/"))
    {
        float m = message.toFloat();
        int x = t.substring(25,26).toInt();
        if (m > -1 && m <= 999999 && x > -1 && x <2) 
        {
            counter[x].value = m;
        }
        publish_message("home/water_count/counter/" + t.substring(25,26),  counter[x].value , true);
    }
}

void setup() {
//Serial.begin(9600);
    WiFi.on();
    WiFi.connect();
    if (waitFor(WiFi.ready, 5000)) {mqtt_connect();}
    for (int i=0; i < 2; i++) 
    {
        pinMode(valve[i].pin, OUTPUT);
        digitalWrite(valve[i].pin, valve[i].state);
        pinMode(counter[i].pin, INPUT);
        pinMode(sensor[i].pin, INPUT);
        counter[i].state = digitalRead(counter[i].pin);
        pinMode(pressure[i], AN_INPUT);
    }
    pinMode(A4, INPUT_PULLUP);

    display.begin(SSD1306_SWITCHCAPVCC, 0x3C);  // initialize with the I2C addr 0x3C (for the 128x64)
    display.clearDisplay();   // clears the screen and buffer

    //Particle.connect();
}

void loop() 
{
    currentMillis = millis();
    // проверяем наличие сети и подключения к MQTT брокеру
    if (currentMillis - previous_conected >= 30000 || previous_conected > currentMillis)
    {
        previous_conected = currentMillis;
        if (!client.isConnected() & wifi_uptime > 60)
        {
            mqtt_connect();
        }
        publish_message("home/water_count/rssi", WiFi.RSSI(), true);
    }
    if (currentMillis - previous_wifi_uptime >= 1000 || previous_wifi_uptime > currentMillis)
    {
        previous_wifi_uptime = currentMillis;
        WiFi.ready() ? wifi_uptime++ : wifi_uptime = 0;
        //work with button and display
        int fg = digitalRead(A4);
        if (display_timeout > 0)
        {
            display_timeout -= 1;
            if (display_timeout == 0) 
            { 
                display.clearDisplay();
                display.display();
            } 
        }
        if (fg == 0)
        {
            if (display_timeout == 0)
            {
                display.clearDisplay();   // clears the screen and buffer
                display.setTextSize(2);
                display.setTextColor(WHITE);
                display.setCursor(0,0);
                display.print("C=");
                display.println(counter[0].value, 4);
                display.setCursor(0,16);
                display.print("H=");
                display.println(counter[1].value, 4);
                display.setCursor(0,32);
                display.print("Valve=");
                display.print(valve[0].state);
                display.print("|");
                display.println(valve[1].state);
                display.setCursor(0,48);
                display.print("Sensor=");
                display.print(sensor[0].state);
                display.print("|");
                display.println(sensor[1].state);
                display.display();
            }
            display_timeout = 10;
        }
    }
    //counter check
    if (currentMillis - previous_counter_read >= 500 || previous_counter_read > currentMillis)
    {
        previous_counter_read = currentMillis;
        for (int i=0; i < 2; i++) 
        {
            byte count_state = digitalRead(counter[i].pin);
            if (count_state != counter[i].state)
            {
                counter[i].state = count_state;
                if (count_state == 0)
                {
                    counter[i].value += 0.01;
                    char buf18[30];
                    sprintf(buf18,"home/water_count/counter/%d", i);
                    publish_message(buf18 , counter[i].value, true);
                }
            }
            // работаем с датчиком протечки
            byte sensor_state = digitalRead(sensor[i].pin);
            if (sensor_state != sensor[i].state) //
            {
                sensor[i].state = sensor_state;
                sensor[i].timeout = SENSOR_TIMEOUT; 
            }
            if (sensor[i].timeout > 0) 
            {
                sensor[i].timeout -= 1;
                if (sensor[i].timeout == 0)
                {
                    char buf18[30];
                    sprintf(buf18,"home/water_count/sensor/%d", i);
                    publish_message(buf18 , sensor[i].state, true);
                    if (sensor[i].state  == 0)
                    {
                        set_valve(0, 1); //close both valve
                        set_valve(1, 1); //close both valve
                    }
                }
            }
        }
    }

    // temp onewire
    if (currentMillis - start_temp_timer >= 299000 || start_temp_timer > currentMillis)
    { //стартуем расчет
        start_temp_timer = currentMillis;
        presense0 = start_temp0();
        presense1 = start_temp1();
    }
    if (currentMillis - read_temp_timer >= 300000 || read_temp_timer > currentMillis)
    {//считывваем темп
        read_temp_timer = currentMillis;
        start_temp_timer = currentMillis;
        if (presense0) read_temp0();
        if (presense1) read_temp1();
        //preasure calc and send
        char buf18[30]; 
        for (int i=0; i < 2; i++)
        {
            sprintf(buf18,"home/water_count/pressure/%d", i);
            float read_val = analogRead(pressure[i]);
            float value = (read_val - 600.0) / 300.0 ;
            publish_message(buf18 , value, false);
        }
    }
    //Particle.process();
    client.loop();
}

void mqtt_connect()
{
    if (client.connect("water_count"))
    { //подпись на spark и публикуем послднее состояние
        client.subscribe("home/water_count/spark/set");
        publish_message("home/water_count/spark", Particle.connected() ? 1 : 0, true);
        client.subscribe("home/water_count/valve/+/set");
        client.subscribe("home/water_count/counter/+/set");
        
    }
}

bool start_temp0()
{
    
    if ( !ds0.search(addr0)) { ds0.reset_search(); return false;}
    ds0.reset_search();
    if (OneWire::crc8(addr0, 7) != addr0[7]) { return false;}
    
    ds0.reset();
    ds0.select(addr0);
    ds0.write(0x44, 0);
    return true; 
}
bool start_temp1()
{
    
    if ( !ds1.search(addr1)) { ds1.reset_search(); return false;}
    ds1.reset_search();
    if (OneWire::crc8(addr1, 7) != addr1[7]) { return false;}
    
    ds1.reset();
    ds1.select(addr1);
    ds1.write(0x44, 0);
    return true; 
}

bool read_temp0()
{
    //delay(1000);
    ds0.reset();
    ds0.select(addr0);
    ds0.write(0xBE, 0);

    for (int i = 0; i < 9; i++) 
    {
        data[i] = ds0.read();
    }
    int16_t raw = (data[1] << 8) | data[0];
    float celsius = (float)raw * 0.0625;
    if (celsius < 0 || celsius > 100) return false;
    publish_message("home/water_count/temp/0", celsius, false);
    //Serial.println(celsius);
    ds0.reset_search();
    return true;
}
bool read_temp1()
{
    //delay(1000);
    ds1.reset();
    ds1.select(addr1);
    ds1.write(0xBE, 0);

    for (int i = 0; i < 9; i++) 
    {
        data[i] = ds1.read();
    }
    int16_t raw = (data[1] << 8) | data[0];
    float celsius = (float)raw * 0.0625;
    if (celsius < 0 || celsius > 100) return false;
    publish_message("home/water_count/temp/1", celsius, false);
    //Serial.println(celsius);
    ds1.reset_search();
    return true;
}

void set_valve(int vlv, byte state)
{
    valve[vlv].state = state;
    digitalWrite(valve[vlv].pin, state);
    char buf26[26];
    sprintf(buf26,"home/water_count/valve/%d", vlv);
    publish_message(buf26 , state , true);
}

Без картинок - если оставить теги картинок, Хабр ломается.
Number	watercount_temp1	"T cool [%.1f °C]"			(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/temp1:state:default]" }
Number	watercount_temp2	"T hot [%.1f °C]"			(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/temp2:state:default]" }
Number	watercount_count0	"Count cool [%.2f М³]"					(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/counter/0:state:default]" }
Number	watercount_count1	"Count hot [%.2f М³]"				(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/counter/1:state:default]" }
Number	watercount_pressure0	"P cool [%.2f Атм.]"			(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/pressure/0:state:default]" }
Number	watercount_pressure1	"P hot [%.2f Атм.]"				(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/pressure/1:state:default]" }
Number	watercount_sensor0	"Sensor0 is [MAP(water_sensor.map):%s]"		(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/sensor/0:state:default]" }
Number	watercount_sensor1	"Sensor1 is [MAP(water_sensor.map):%s]"		(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/sensor/1:state:default]" }
Number	watercount_valve0	"Valve cool"		(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/valve/0:state:default], >[mqtt_bro:home/water_count/valve/0/set:command:*:default]" }
Number	watercount_valve1	"Valve hot"				(gWaterCount)		{ mqtt="<[mqtt_bro:home/water_count/valve/1:state:default], >[mqtt_bro:home/water_count/valve/1/set:command:*:default]" }
String	watercount_sendStr "LastVol:[%s]"						(gWaterCount)
Number	watercount_sendCool "Send cool [%.2f М³]"		(gWaterCount)
Number	watercount_sendHot "Send hot [%.2f М³]"			(gWaterCount)
Number	watercount_sendSwitch "Autosend" 	(gWaterCount)
Number	watercount_rssi	"WaterCount [%d dB]"	(gSpark_RSSI)		{ mqtt="<[mqtt_bro:home/water_count/rssi:state:default]" }
Number	watercount_spark_state	"WaterCount Spark"		(gSpark)		{ mqtt="<[mqtt_bro:home/water_count/spark:state:default], >[mqtt_bro:home/water_count/spark/set:command:*:default]" }

rule "Check watercount_sensor0"
	when
		Item watercount_sensor0 received update
	then
		if ((watercount_sensor0.state as DecimalType) == 1)
		{
			if ((watercount_sensor0.historicState(now.minusSeconds(3)).state as DecimalType) == 1)
			{
				sendTelegram("****_bot", "Sensor0 was wet less than 5 seconds")
			}
			else
			{
				sendTelegram("****_bot", "Sensor0 become dry")
			}
		}
		else
		{
			if ((watercount_sensor0.historicState(now.minusSeconds(3)).state as DecimalType) == 0)
			{
				sendTelegram("****_bot", "Sensor0 was dry less than 5 seconds");			
			}
			else
			{
				sendTelegram("****_bot", "Sensor0 become wet! Valves will be closed!")
			}
		}
	end
rule "Check watercount_sensor1"
	when
		Item watercount_sensor1 received update
	then
		if ((watercount_sensor1.state as DecimalType) == 1)
		{
			if ((watercount_sensor1.historicState(now.minusSeconds(3)).state as DecimalType) == 1)
			{
				sendTelegram("****_bot", "Sensor1 was wet less than 5 seconds")
			}
			else
			{
				sendTelegram("****_bot", "Sensor1 become dry")
			}
		}
		else
		{
			if ((watercount_sensor1.historicState(now.minusSeconds(3)).state as DecimalType) == 0)
			{
				sendTelegram("****_bot", "Sensor1 was dry less than 5 seconds");			
			}
			else
			{
				sendTelegram("****_bot", "Sensor1 become wet! Valves will be closed!")
			}
		}
	end
	
rule "Check watercount_temp2"
	when
		Item watercount_temp2 received update
	then
		if ((watercount_temp2.state as DecimalType) < 37 )
		{
			sendTelegram("****_bot", String::format("Hot water temp drop to %s", watercount_temp2.state.toString));
		}
	end

rule "Check watercount_pressure0"
	when
		Item watercount_pressure0 received update
	then
		if ((watercount_pressure0.state as DecimalType) < 1 && (watercount_pressure0.historicState(now.minusSeconds(3)).state as DecimalType) >= 1)
		{
			sendTelegram("****_bot", String::format("Cool pressure drop to %s", watercount_pressure0.state.toString));
		}
		if ((watercount_pressure0.state as DecimalType) > 1 && (watercount_pressure0.historicState(now.minusSeconds(3)).state as DecimalType) <= 1)
		{
			sendTelegram("****_bot", String::format("Cool pressure rise to %s", watercount_pressure0.state.toString));
		}
	end
rule "Check watercount_pressure1"
	when
		Item watercount_pressure1 received update
	then
		if ((watercount_pressure1.state as DecimalType) < 1 && (watercount_pressure1.historicState(now.minusSeconds(3)).state as DecimalType) >= 1)
		{
			sendTelegram("****_bot", String::format("Hot pressure drop to %s", watercount_pressure1.state.toString));
		}
		if ((watercount_pressure1.state as DecimalType) > 1 && (watercount_pressure1.historicState(now.minusSeconds(3)).state as DecimalType) <= 1)
		{
			sendTelegram("****_bot", String::format("Hot pressure rise to %s", watercount_pressure1.state.toString));
		}
	end

rule "Generate send string counters" //every 24 day of mounth  in 00.01 minutes
	when
		Time cron "0 0 1 24 1/1 ?" 
	then
		
		var float deltaCool = (watercount_count0.state as DecimalType).floatValue() - (watercount_sendCool.state as DecimalType).floatValue()
		var float deltaHot = (watercount_count1.state as DecimalType).floatValue() - (watercount_sendHot.state as DecimalType).floatValue()
		
		if (deltaCool >= 0 && deltaHot >= 0)
		{
			watercount_sendStr.postUpdate(String::format(" %.2f / %.2f м3", deltaCool, deltaHot))
			watercount_sendCool.state = watercount_count0.state
			watercount_sendHot.state = watercount_count1.state
			sendTelegram("****_bot", String::format("Лизюкова 23, корп 5, кв. 23. Счетчик №2560097 (хол.вода) = %.2f м3. Cчетчик №2538996 (гор.вода) = %.2f м3. %s", (watercount_sendCool.state as DecimalType).floatValue(), (watercount_sendHot.state as DecimalType).floatValue(), watercount_sendStr.state.toString()))
		}
		else
		{
			watercount_sendSwitch.postUpdate(0)
			sendTelegram("****_bot", "Current counters value less than sended last time. Turn off autosend.")
		}
end
rule "Send string counters" 
	when
		Time cron "0 0 23 24 1/1 ?"
	then
		if (watercount_sendSwitch.state == 1)
		{
			sendMail("uk@uk.ru", "Лизюкова 23, корп 5, кв. 23", String::format("Лизюкова 23, корп 5, кв. 23. Счетчик №2560097 (хол.вода) = %.2f м3. Cчетчик №2538996 (гор.вода) = %.2f м3", (watercount_sendCool.state as DecimalType).floatValue(), (watercount_sendHot.state as DecimalType).floatValue()));
			sendTelegram("****_bot", "Send email with watercount values");
		}
		else
		{
			sendTelegram("****_bot", "Can't send email with watercount values - autosend is OFF.");
		}
end
rule "Rotate valves" 
	when
		Time cron "0 0 05 25 1/1 ?"
	then
		if (watercount_valve0.state == 0 && watercount_valve1.state == 0)
		{	
			watercount_valve0.postUpdate(1)
			Thread::sleep(1000)
			watercount_valve1.postUpdate(1)
			Thread::sleep(1000)
			watercount_valve0.postUpdate(0)
			Thread::sleep(1000)
			watercount_valve1.postUpdate(0)
			sendTelegram("****_bot", "Valves was rotated.");
		}
		else
		{
			sendTelegram("****_bot", "Can't rotate valves, it's closed.");
		}
end