Micropython code for monitoring PH, ORP, and temperature of your hot tub using ESP32, Home Assistant, and InfluxDB2
# MIT License (MIT)
# Copyright (c) 2023 Stephen Carey
# https://opensource.org/licenses/MIT

# OTA:file:display_handler.py
# OTA:reboot:false
import ssd1306

from machine import Pin, SoftI2C
import uasyncio as asyncio

TEXT_ROW_HEIGHT = 9


class DisplayHandler:
    def __init__(self, sda_pin_num, scl_pin_num, on_off_pin_num, num_calibration_readings, row_height=TEXT_ROW_HEIGHT):
        self.on_off_pin = Pin(on_off_pin_num, Pin.IN, pull=Pin.PULL_UP)
        self.row_height = row_height
        self.oled = None
        self.i2c = SoftI2C(sda=Pin(sda_pin_num), scl=Pin(scl_pin_num), freq=100000)

        self.command = None
        self.data = None
        self.temp_unit = 'F'
        self.new_neutral_voltages = None
        self.new_acid_voltages = None
        self.num_calibration_readings = num_calibration_readings

    async def init(self):
        print(f'I2C scan results: {self.i2c.scan()}')
        try:
            self.oled = ssd1306.SSD1306_I2C(128, 64, self.i2c)

            # startup tests
            self.oled.fill(1)
            self.oled.show()
            await asyncio.sleep(1)
            print("OLED init success")
        except Exception as e:
            print("OLED init failed: {}".format(e))

    def is_display_on(self):
        return self.on_off_pin.value() == 0

    def update_values(self, command, data, new_neutral_voltages, new_acid_voltages,
                      temp_unit):
        self.command = command
        self.data = data
        self.temp_unit = temp_unit
        self.new_neutral_voltages = new_neutral_voltages
        self.new_acid_voltages = new_acid_voltages

    def show_rows(self, rows):
        self.oled.fill(0)
        for i in range(len(rows)):
            self.oled.text(rows[i], 0, self.row_height * i, 1)
        self.oled.show()

    async def show_display(self):
        while True:
            if self.oled:
                self.oled.fill(0)
                if not self.command:
                    if self.is_display_on() and self.data.get('air_temp', None):
                        rows = [
                            f'Air temp: {round(self.data["air_temp"], 1)}{self.temp_unit}',
                            f'pH:       {round(self.data["ph"], 1)}',
                            f'ORP:      {round(self.data["orp"])}mV',
                            f'H2O temp: {round(self.data["water_temp"], 1)}{self.temp_unit}'
                        ]
                        self.show_rows(rows)
                    await asyncio.sleep(5)
                else:  # calibrate
                    if self.is_display_on():
                        num_acid_readings = len(self.new_acid_voltages)
                        num_neutral_readings = len(self.new_neutral_voltages)
                        rows = [
                            "Calibrating...",
                            f'ph4 count: {num_acid_readings}/{self.num_calibration_readings}',
                            f'ph4 avg: {"N/A" if num_acid_readings == 0 else sum(self.new_acid_voltages) / num_acid_readings}mV',
                            f'ph7 count: {len(self.new_neutral_voltages)}/{self.num_calibration_readings}',
                            f'ph7 avg: {"N/A" if num_neutral_readings == 0 else sum(self.new_neutral_voltages) / num_neutral_readings}mV'
                        ]
                        self.show_rows(rows)
                    await asyncio.sleep(1)
                self.oled.show()
            else:
                await asyncio.sleep(1)

display_handler.py
scarey/hot-tub-monitor
# MIT License (MIT)
# Copyright (c) 2023 Stephen Carey
# https://opensource.org/licenses/MIT

# OTA:file:ha.py
# OTA:reboot:true
import binascii
import json
import uos
from machine import unique_id
from tub_config import TubConfig


async def setup_ha_discovery(config_data: TubConfig, version):
    identifier = binascii.hexlify(unique_id()).decode()
    device = {
        "identifiers": identifier,
        "name": "DIY Hot Tub Monitor",
        "sw_version": f"{uos.uname().version}_{version}",
        "model": uos.uname().machine,
        "manufacturer": identifier
    }

    topic = "homeassistant/sensor/hottub/airtemp/config"
    message = {
        "name": "Local Temperature",
        "availability_topic": config_data.available_topic,
        "qos": 0,
        "device_class": "temperature",
        "device": device,
        "state_topic": "esp32/hottub/readings",
        "value_template": "{{ value_json.air_temp }}",
        "unit_of_measurement": f"°{config_data.temp_unit}",
        "unique_id": "local-air-temperature",
        'force_update': True
    }
    await config_data.client.publish(topic, json.dumps(message).encode("UTF-8"), True)

    topic = "homeassistant/sensor/hottub/temp/config"
    message = {
        "name": "Hot Tub Temperature",
        "availability_topic": config_data.available_topic,
        "qos": 0,
        "device_class": "temperature",
        "device": device,
        "state_topic": "esp32/hottub/readings",
        "value_template": "{{ value_json.water_temp }}",
        "unit_of_measurement": f"°{config_data.temp_unit}",
        "unique_id": "hottub-water-temperature",
        'force_update': True
    }
    await config_data.client.publish(topic, json.dumps(message).encode("UTF-8"), True)

    topic = "homeassistant/sensor/hottub/ph/config"
    message = {
        "name": "Hot Tub pH",
        "availability_topic": config_data.available_topic,
        "qos": 0,
        "icon": "mdi:ph",
        "device": device,
        "state_topic": "esp32/hottub/readings",
        "value_template": "{{ value_json.ph }}",
        "unit_of_measurement": "pH",
        "unique_id": "hottub-water-ph",
        'force_update': True
    }
    await config_data.client.publish(topic, json.dumps(message).encode("UTF-8"), True)

    topic = "homeassistant/sensor/hottub/orp/config"
    message = {
        "name": "Hot Tub ORP",
        "availability_topic": config_data.available_topic,
        "qos": 0,
        "icon": "mdi:react",
        "device": device,
        "state_topic": "esp32/hottub/readings",
        "value_template": "{{ value_json.orp }}",
        "unit_of_measurement": "mV",
        "unique_id": "hottub-water-orp",
        'force_update': True
    }
    await config_data.client.publish(topic, json.dumps(message).encode("UTF-8"), True)

    topic = "homeassistant/button/hottub/command/config"
    message = {
        "name": "Hot Tub calibrate",
        "availability_topic": config_data.available_topic,
        "qos": 0,
        "icon": "mdi:ruler-square-compass",
        "device": device,
        "command_topic": "esp32/hottub/command",
        "command_template": "{\"command\": \"calibrate\",\"timestamp\": \"{{ this.last_changed }}\"}",
        "unique_id": "hottub-calibrate",
    }
    await config_data.client.publish(topic, json.dumps(message).encode("UTF-8"), True)

ha.py
scarey/hot-tub-monitor
# MIT License (MIT)
# Copyright (c) 2023 Stephen Carey
# https://opensource.org/licenses/MIT

# OTA:file:main.py
# OTA:reboot:true
import binascii
import ds18x20
import json
import onewire
import sys
import uasyncio as asyncio

from machine import ADC, Pin
from display_handler import DisplayHandler
from tub_config import TubConfig

BASE_TOPIC = 'esp32/hottub'
COMMAND_TOPIC = f'{BASE_TOPIC}/command'
AVAILABLE_TOPIC = f'{BASE_TOPIC}/availability'

NUM_CALIBRATION_READINGS = 42
VERSION = 3

air_temp = 0
water_temp = 0
ph_val = 0
orp_val = 0
new_neutral_voltages = []
new_acid_voltages = []

# ph is 0V to 3.0V in
ph_pin = ADC(Pin(35))
ph_pin.atten(ADC.ATTN_11DB)
# orp is -2V to +2V in though we should never see anything negative for this use
orp_pin = ADC(Pin(34))
orp_pin.atten(ADC.ATTN_11DB)

temp_pin = Pin(4, Pin.IN, pull=Pin.PULL_UP)
cover_state_pin = Pin(5, Pin.IN, pull=Pin.PULL_UP)

ds = ds18x20.DS18X20(onewire.OneWire(temp_pin))
tc = TubConfig(BASE_TOPIC, ds)
display_handler = DisplayHandler(sda_pin_num=21, scl_pin_num=22, on_off_pin_num=26,
                                 num_calibration_readings=NUM_CALIBRATION_READINGS)


def read_ph_mv():
    return ph_pin.read_uv() / 1000


def read_average(num_readings, read_func):
    total = 0
    for _ in range(num_readings):
        total += read_func()
    return total / num_readings


def read_ph():
    milli_voltage = read_ph_mv()
    calibrated_neutral_v = tc.config['ph_neutral_calibration']
    calibrated_acid_v = tc.config['ph_acid_calibration']

    # https://github.com/DFRobot/DFRobot_PH/blob/master/DFRobot_PH.cpp
    # can't find any temp adjustment data but things seem to work well w/o it at hot tub temps
    slope = (7.0 - 4.0) / ((calibrated_neutral_v - 1500.0) / 3.0 - (calibrated_acid_v - 1500.0) / 3.0)
    intercept = 7.0 - slope * (calibrated_neutral_v - 1500.0) / 3.0

    ph_value = slope * (milli_voltage - 1500.0) / 3.0 + intercept  # y = k*x + b

    print(f'Adjusted PH: {ph_value}, raw V: {milli_voltage / 1000}')
    return ph_value


def read_orp():
    voltage = orp_pin.read_uv() / 1_000_000
    orp_mv = (2.5 - voltage) / 1.037 * 1000.0  # * 1000 to get mV
    print(f'ORP mV: {orp_mv}, raw V: {voltage}')
    return orp_mv


async def read_temps():
    global water_temp, air_temp
    temp_error = ''

    if len(tc.roms) != 2:
        tc.init_roms()
    if len(tc.roms) > 0:
        ds.convert_temp()
        await asyncio.sleep_ms(750)
        for rom in tc.expected_roms:
            try:
                if rom in tc.roms:
                    current_temp = ds.read_temp(binascii.unhexlify(rom))
                    if tc.temp_unit == 'F':
                        current_temp = current_temp * 1.8 + 32

                    rounded_temp = round(current_temp, 1)
                    if rom == tc.air_rom:
                        air_temp = rounded_temp
                    else:
                        water_temp = rounded_temp
                else:
                    temp_error = f'Missing expected rom: {rom}, these were detected: {tc.roms}'
                await asyncio.sleep(1)
            except Exception as e:
                temp_error = str(e)
                print(f'Problem reading temp: {e}')
    return water_temp, air_temp, temp_error


# The calibration process takes PH readings every second until it has multiple readings for PH 4
# and PH 7.  It then takes the average of the last few readings of each PH and updates the config
# with those calibration values.
async def calibrate():
    print("Calibration starting...")
    new_acid_voltages.clear()
    new_neutral_voltages.clear()
    while len(new_neutral_voltages) != NUM_CALIBRATION_READINGS or len(new_acid_voltages) != NUM_CALIBRATION_READINGS:
        # let's get 42 readings in each range, average the last 12 and send the new calibration data
        read_voltage = read_ph_mv()

        if 1322 < read_voltage < 1678:
            if len(new_neutral_voltages) < NUM_CALIBRATION_READINGS:
                new_neutral_voltages.append(read_voltage)
        elif 1854 < read_voltage < 2210:
            if len(new_acid_voltages) < NUM_CALIBRATION_READINGS:
                new_acid_voltages.append(read_voltage)

        display_handler.update_values(tc.command, {}, new_neutral_voltages,
                                      new_acid_voltages, tc.temp_unit)
        await asyncio.sleep(1)
        print(f'New neutral readings: {len(new_neutral_voltages)}, new acid readings: {len(new_acid_voltages)}')

    # don't include first 30 readings in average
    average_neutral_voltage = sum(new_neutral_voltages[30:]) / (NUM_CALIBRATION_READINGS - 30)
    average_acid_voltage = sum(new_acid_voltages[30:]) / (NUM_CALIBRATION_READINGS - 30)

    # backup existing config
    await tc.backup_config()
    # write new config
    tc.config['ph_neutral_calibration'] = average_neutral_voltage
    tc.config['ph_acid_calibration'] = average_acid_voltage
    try:
        command_dict = json.loads(tc.command)
        tc.config['last_calibration'] = command_dict.get('timestamp', 'Unknown')
    except Exception as e:
        print(f"Problem parsing command json: {tc.command}")
        sys.print_exception(e)
    await tc.publish_config()
    print("Calibration complete!")


async def main():
    await tc.client.connect()
    await asyncio.sleep(2)  # Give broker time
    await tc.online()
    await display_handler.init()
    try:
        import ha
        print("ha module found, configuring HA discovery.")
        await ha.setup_ha_discovery(tc, VERSION)
    except ImportError:
        print("ha module not found, not configuring HA discovery.")
    await tc.client.publish(tc.version_topic, str(VERSION), retain=True)
    global air_temp, water_temp, ph_val, orp_val
    while True:
        try:
            while not tc.config_done:
                print("Config not found, will check again in a few secs...")
                await asyncio.sleep(5)
            if tc.command:
                await calibrate()
                tc.command = None
            else:
                ph_val = read_average(5, read_ph)
                orp_val = read_average(5, read_orp)
                air_temp, water_temp, temp_error = await read_temps()
                new_data = {}
                new_data['ph'] = ph_val
                new_data['orp'] = orp_val
                new_data['water_temp'] = water_temp
                new_data['air_temp'] = air_temp
                new_data['temp_unit'] = tc.temp_unit
                new_data['temp_error'] = temp_error
                new_data['cover_state'] = cover_state_pin.value()
                display_handler.update_values(tc.command, new_data, [], [], tc.temp_unit)
                print(f"Publishing updated data: {new_data}")
                await tc.client.publish(tc.readings_topic, json.dumps(new_data), True)
                await asyncio.sleep(60)
        except Exception as e:
            print("Problem in main loop: ", e)
            await asyncio.sleep(5)


try:
    loop = asyncio.get_event_loop()
    loop.create_task(main())
    loop.create_task(display_handler.show_display())
    loop.run_forever()
finally:
    tc.client.close()
    asyncio.stop()

main.py
scarey/hot-tub-monitor
# MIT License (MIT)
# Copyright (c) 2023 Stephen Carey
# https://opensource.org/licenses/MIT

# OTA:file:tub_config.py
# OTA:reboot:true
import binascii
import json
import sys
import uasyncio as asyncio
from mqtt_local import config
from mqtt_as import MQTTClient


class TubConfig:
    def __init__(self, base_topic, ds):
        self.base_topic = base_topic
        self.ds = ds
        self.version_topic = f'{base_topic}/version'
        self.ota_topic = f'{base_topic}/ota'
        self.command_topic = f'{base_topic}/command'
        self.config_topic = f'{base_topic}/config'
        self.readings_topic = f'{base_topic}/readings'
        self.available_topic = f'{base_topic}/availability'

        config['subs_cb'] = self.handle_incoming_message
        config['connect_coro'] = self.conn_han
        config['wifi_coro'] = self.wifi_han
        config['will'] = [self.available_topic, 'offline', True, 0]

        MQTTClient.DEBUG = False
        self.client = MQTTClient(config)
        self.command = None
        self.config_done = False
        self.temp_unit = None
        self.config = {}
        self.expected_roms = []
        self.air_rom = None
        self.water_rom = None
        self.roms = []

    async def backup_config(self):
        json_data = json.dumps(self.config)
        print(f'Publishing config backup: {json_data}')
        await self.client.publish(f'{self.config_topic}bak', json_data, True)

    async def publish_config(self):
        json_data = json.dumps(self.config)
        print(f'Publishing updated config: {json_data}')
        await self.client.publish(self.config_topic, json_data, True)

    def handle_incoming_message(self, topic, msg, retained):
        topic_string = str(topic, 'UTF-8')
        msg_string = str(msg, 'UTF-8')
        if len(msg_string) < 500:
            print(f'{topic_string}: {msg_string}')
        else:
            print(f'Got a big message on {topic_string}...')
        if topic == self.config_topic:
            try:
                self.config = json.loads(msg_string)
                if 'ph_neutral_calibration' not in self.config or 'ph_acid_calibration' not in self.config:
                    print(
                        'Values for "ph_neutral_calibration" and "ph_acid_calibration" must be specified.  Good defaults are 1500.0 and 2032.44 respectively')
                elif 'air_rom_reg_num' not in self.config or 'water_rom_reg_num' not in self.config:
                    print(
                        'Values for "air_rom_reg_num" and "water_rom_reg_num" must be specified.  The hex values should be printed during startup.')
                else:
                    self.temp_unit = 'F' if self.config.get('temp_unit', 'F') == 'F' else 'C'
                    self.expected_roms.clear()
                    self.air_rom = self.config['air_rom_reg_num']
                    self.expected_roms.append(self.air_rom)
                    self.water_rom = self.config['water_rom_reg_num']
                    self.expected_roms.append(self.water_rom)
                    self.init_roms()
                    self.config_done = True
            except Exception as e:
                print(f'Problem with tub config: {e}')
                sys.print_exception(e)
        elif topic == self.command_topic:
            self.command = msg_string
        elif topic == self.ota_topic:
            import ota
            ota.process_ota_msg(msg_string)

    def init_roms(self):
        self.roms = [binascii.hexlify(x).decode() for x in self.ds.scan()]
        print("Scan found these ROMS:")
        for rom in self.roms:
            print(f'ROM registration number hex: {rom}')

    async def wifi_han(self, state):
        print('Wifi is ', 'up' if state else 'down')
        await asyncio.sleep(1)

    # If you connect with clean_session True, must re-subscribe (MQTT spec 3.1.2.4)
    async def conn_han(self, client):
        await client.subscribe(self.ota_topic, 0)
        await client.subscribe(self.command_topic, 0)
        await client.subscribe(self.config_topic, 0)
        await self.online()

    async def online(self):
        await self.client.publish(self.available_topic, 'online', retain=True, qos=0)

tub_config.py
scarey/hot-tub-monitor