1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280

# SPDX-FileCopyrightText: 2025 John Park for Adafruit Industries
#
# SPDX-License-Identifier: MIT
'''
Espresso Tank Meter
Feather ESP32-S2 with RCWL-1601 Ultrasonic distance sensor
'''

import time
import os
import ssl
import microcontroller
import supervisor
import socketpool
import wifi
import board
import digitalio
import alarm
import neopixel
import adafruit_hcsr04
import adafruit_minimqtt.adafruit_minimqtt as MQTT
from adafruit_io.adafruit_io import IO_MQTT
import adafruit_requests
import adafruit_max1704x

# Initialize the power pin for the sensor
sensor_power = digitalio.DigitalInOut(board.A2)
sensor_power.direction = digitalio.Direction.OUTPUT
sensor_power.value = False  # Start with sensor powered off

def power_sensor_on():
    """Turn on power to the ultrasonic sensor and wait for it to stabilize."""
    sensor_power.value = True
    time.sleep(0.55)  # Give sensor time to power up and stabilize

def power_sensor_off():
    """Turn off power to the ultrasonic sensor."""
    sensor_power.value = False

# Initialize the sonar sensor
sonar = adafruit_hcsr04.HCSR04(trigger_pin=board.A0, echo_pin=board.A1)

# Initialize the battery monitor
i2c = board.I2C()  # uses board.SCL and board.SDA
battery_monitor = adafruit_max1704x.MAX17048(i2c)

# Define colors (hex values)
WHITE = 0xFFFFFF
BLUE = 0x0000FF
GREEN = 0x00FF00
YELLOW = 0xFFFF00
RED = 0xFF0000
PINK = 0xbb00bb
CYAN = 0x00bbbb
OFF = 0x000000

# Initialize the NeoPixel
pixel = neopixel.NeoPixel(board.NEOPIXEL, 1, brightness=0.25)
# Show yellow on startup
pixel.fill(YELLOW)

# Operating hours (24-hour format with minutes, e.g., "6:35" and "16:00")
OPENING_TIME = "6:00"
CLOSING_TIME = "15:30"
# Normal operation check interval
NORMAL_CHECK_MINUTES = 10
# Sleep duration in seconds during operating hours
SLEEP_DURATION = 60 * NORMAL_CHECK_MINUTES
# Display duration in seconds
DISPLAY_DURATION = 1
# Number of samples to average
NUM_SAMPLES = 5

def parse_time(time_str):
    """Convert time string (HH:MM format) to hours and minutes."""
    # pylint: disable=redefined-outer-name
    parts = time_str.split(':')
    return int(parts[0]), int(parts[1])

def get_average_distance():
    """Take multiple distance readings and return the average."""
    power_sensor_on()  # Power on the sensor before taking measurements
    distances = []
    for _ in range(NUM_SAMPLES):
        try:
            distance = sonar.distance
            distances.append(distance)
            time.sleep(0.1)  # Short delay between readings
        except RuntimeError:
            print("Error reading distance")
            continue
    power_sensor_off()  # Power off the sensor after measurements

    # Only average valid readings
    if distances:
        return sum(distances) / len(distances)
    return None

def set_pixel_color(distance):
    """Set NeoPixel color based on distance."""
    if distance is None:
        pixel.fill(OFF)
        return

    if distance < 2:
        pixel.fill(WHITE)
    elif 2 <= distance < 10:
        pixel.fill(BLUE)
    elif 10 <= distance < 16:
        pixel.fill(GREEN)
    elif 18 <= distance < 20:
        pixel.fill(YELLOW)
    else:  # distance >= 22
        pixel.fill(RED)

# Wait for things to settle before reading sonar
time.sleep(0.1)

# Get average distance
avg_distance = get_average_distance()

if avg_distance is not None:

    if avg_distance >= 22:
        # pylint: disable=invalid-name
        avg_distance = 22
    print(f"Average distance: {avg_distance:.1f} cm")
    # Set color based on average distance
    set_pixel_color(avg_distance)

    # Check battery status
    battery_voltage = battery_monitor.cell_voltage
    battery_percent = battery_monitor.cell_percent
    print(f"Battery: {battery_percent:.1f}% ({battery_voltage:.2f}V)")

    # Try connecting to WiFi
    try:

        print("Connecting to %s" % os.getenv("CIRCUITPY_WIFI_SSID"))
        # Show pink while attempting to connect
        pixel.fill(PINK)
        wifi.radio.connect(os.getenv("CIRCUITPY_WIFI_SSID"), os.getenv("CIRCUITPY_WIFI_PASSWORD"))
        print("Connected to %s" % os.getenv("CIRCUITPY_WIFI_SSID"))
        # Show cyan on successful connection
        pixel.fill(CYAN)
        time.sleep(1)  # Brief pause to show the connection success
    # pylint: disable=broad-except
    except Exception as e:
        print("Failed to connect to WiFi. Error:", e, "\nBoard will hard reset in 30 seconds.")
        pixel.fill(OFF)
        time.sleep(10)
        microcontroller.reset()

    # Create a socket pool
    pool = socketpool.SocketPool(wifi.radio)
    requests = adafruit_requests.Session(pool, ssl.create_default_context())

    # Initialize a new MQTT Client object
    mqtt_client = MQTT.MQTT(
        broker="io.adafruit.com",
        username=os.getenv("ADAFRUIT_AIO_USERNAME"),
        password=os.getenv("ADAFRUIT_AIO_KEY"),
        socket_pool=pool,
        ssl_context=ssl.create_default_context(),
    )

    # Initialize Adafruit IO MQTT "helper"
    io = IO_MQTT(mqtt_client)

    try:
        # If Adafruit IO is not connected...
        if not io.is_connected:
            print("Connecting to Adafruit IO...")
            io.connect()

        # Get current time from AIO time service
        aio_username = os.getenv("ADAFRUIT_AIO_USERNAME")
        aio_key = os.getenv("ADAFRUIT_AIO_KEY")
        timezone = os.getenv("TIMEZONE")
        # pylint: disable=line-too-long
        TIME_URL = f"https://io.adafruit.com/api/v2/{aio_username}/integrations/time/strftime?x-aio-key={aio_key}&tz={timezone}"
        TIME_URL += "&fmt=%25Y-%25m-%25d+%25H%3A%25M%3A%25S.%25L+%25j+%25u+%25z+%25Z"

        print("Getting time from Adafruit IO...")
        response = requests.get(TIME_URL)
        time_str = response.text.strip()  # Remove any leading/trailing whitespace
        print("Current time:", time_str)

        # Parse the current time from the time string
        time_parts = time_str.split()
        current_time = time_parts[1].split(':')
        current_hour = int(current_time[0])
        current_minute = int(current_time[1])

        # Get opening and closing hours and minutes
        opening_hour, opening_minute = parse_time(OPENING_TIME)
        closing_hour, closing_minute = parse_time(CLOSING_TIME)

        # Convert all times to minutes for easier comparison
        current_minutes = current_hour * 60 + current_minute
        opening_minutes = opening_hour * 60 + opening_minute
        closing_minutes = closing_hour * 60 + closing_minute

        # Check if we're within operating hours
        if opening_minutes <= current_minutes < closing_minutes:
            print(f"Within operating hours ({OPENING_TIME} to {CLOSING_TIME}), proceeding with measurement")

            # Explicitly pump the message loop
            io.loop()

            # Send the distance data
            print(f"Publishing {avg_distance:.1f} to espresso water level feed")
            io.publish("espresso-water-tank-level", f"{avg_distance:.1f}")

            # Send the battery data
            print(f"Publishing {battery_percent:.1f} to battery level feed")
            io.publish("espresso-water-sensor-battery", f"{battery_percent:.1f}")


            # Make sure the message gets sent
            io.loop()

            print("Water level sent successfully")

            # Keep NeoPixel lit for DISPLAY_DURATION seconds
            time.sleep(DISPLAY_DURATION)

            # Use normal check interval during operating hours
            # # pylint: disable=invalid-name
            sleep_seconds = SLEEP_DURATION
            print(f"Next check in {NORMAL_CHECK_MINUTES} minutes")
        else:
            print(f"Outside operating hours ({OPENING_TIME} to {CLOSING_TIME}), going back to sleep")
            # Calculate time until next opening
            if current_minutes >= closing_minutes:
                # After closing, calculate time until opening tomorrow
                minutes_until_open = (24 * 60 - current_minutes) + opening_minutes
            else:
                # Before opening, calculate time until opening today
                minutes_until_open = opening_minutes - current_minutes

            # Convert minutes to seconds for sleep duration
            sleep_seconds = minutes_until_open * 60
            hours_until_open = minutes_until_open // 60
            minutes_remaining = minutes_until_open % 60
            if minutes_remaining:
                print(f"Sleeping until {OPENING_TIME} ({hours_until_open} hours, {minutes_remaining} minutes)")
            else:
                print(f"Sleeping until {OPENING_TIME} ({hours_until_open} hours)")

        response.close()

    # pylint: disable=broad-except
    except Exception as e:
        print("Failed to get or send data, or connect. Error:", e,
              "\nBoard will hard reset in 30 seconds.")
        pixel.fill(OFF)
        time.sleep(30)
        microcontroller.reset()

else:
    print("Failed to get valid distance readings")
    pixel.fill(OFF)
    # pylint: disable=invalid-name
    sleep_seconds = SLEEP_DURATION  # Use normal interval if we couldn't get readings

# Prepare for deep sleep
power_sensor_off()  # Make sure sensor is powered off before sleep
pixel.brightness = 0  # Turn off NeoPixel

# Flush the serial output before sleep
# pylint: disable=pointless-statement
supervisor.runtime.serial_bytes_available
time.sleep(0.05)

# Create time alarm
time_alarm = alarm.time.TimeAlarm(monotonic_time=time.monotonic() + sleep_seconds)

# Enter deep sleep
alarm.exit_and_deep_sleep_until_alarms(time_alarm)