commit 558ee43149b4c757449bd5f38c3b17446e6984a0 Author: Adolfo Reyna Date: Sun May 31 22:44:49 2026 -0400 Initial commit: ESP32-S3-RLCD-4.2 hardware utility classes and demo diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f2aa58b --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +# Temporary generated media and screenshots +*.pbm +*.png +*.pcm +*.wav + +# Python temporary files +__pycache__/ +*.pyc + +# OS generated files +.DS_Store diff --git a/audio_util.py b/audio_util.py new file mode 100644 index 0000000..f7073a9 --- /dev/null +++ b/audio_util.py @@ -0,0 +1,128 @@ +import time +import machine +from machine import Pin, I2C, I2S + +class ES7210: + """MicroPython driver for the ES7210 4-Channel Audio ADC (Microphone Array). + + Controls the ES7210 chip over I2C to configure clocks, channels, gain, and format. + """ + ADDR = 0x40 + + def __init__(self, i2c): + self.i2c = i2c + + def init(self, sample_rate=16000, bit_width=16): + """Initializes the ES7210 registers for dual-microphone recording. + + Args: + sample_rate (int): Audio sample rate (e.g. 16000, 44100, 48000). + bit_width (int): Data bit depth (16 or 24). + + Returns: + bool: True if initialization was successful, False otherwise. + """ + print("Initializing ES7210 Microphone ADC...") + try: + # 1. Reset the chip + self._write(0x00, 0xFF) # Write all 1s to reset register + time.sleep_ms(10) + self._write(0x00, 0x00) # Release reset + + # 2. Power management and system configuration + self._write(0x01, 0x00) # Enable analog power, reference voltage + self._write(0x11, 0x60) # Enable master clock PLL + + # 3. Configure Clock Dividers + if sample_rate == 16000: + self._write(0x02, 0x0C) # BCLK divider + self._write(0x03, 0x10) # LRCK divider + else: # 44100 / 48000 defaults + self._write(0x02, 0x04) + self._write(0x03, 0x08) + + # 4. Input Configuration (Enable Mics 1 and 2, power down Mics 3 and 4) + self._write(0x47, 0x00) # Enable MIC1 / MIC2 analog front-ends + self._write(0x48, 0xFF) # Power down MIC3 / MIC4 path + self._write(0x49, 0x0A) # Power up PGA (Programmable Gain Amplifier) 1 and 2 + self._write(0x4A, 0x00) # Power down PGA 3 and 4 + + # 5. Microphone Gain Settings (+24dB standard) + # Gain range: 0x00 (0dB) to 0x0F (+45dB) in 3dB steps. 0x08 = +24dB. + self._write(0x43, 0x08) # Set MIC1 Gain (+24dB) + self._write(0x44, 0x08) # Set MIC2 Gain (+24dB) + + # 6. Set Digital Interface Format (I2S standard format) + # Bit width: 0x00 = 24-bit, 0x01 = 16-bit, 0x02 = 8-bit, 0x03 = 32-bit + fmt = 0x01 if bit_width == 16 else 0x00 + self._write(0x13, fmt) # Set serial output interface format + self._write(0x14, 0x18) # Enable frame clock / bit clock output + + # 7. Unmute ADCs and enable output + self._write(0x12, 0x00) # Enable ADC digital filters (unmute) + self._write(0x15, 0x30) # Enable output data pin (SDOUT) active + + print("ES7210 initialization complete.") + return True + except Exception as e: + print(f"Failed to initialize ES7210: {e}") + return False + + def _write(self, reg, val): + self.i2c.writeto_mem(self.ADDR, reg, bytes([val])) + + +def record_audio(duration_seconds=5, filename='recording.pcm'): + """Records raw stereo PCM data from the dual microphones to a file. + + Args: + duration_seconds (int): How long to record in seconds. + filename (str): Name of output raw PCM file on the device. + """ + # 1. Start I2C Control Bus (SDA=13, SCL=14) + i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + + # 2. Configure I2S Receiver + # Pins: sck=BCLK (GPIO 9), ws=WS/LRCK (GPIO 45), sd=DIN (GPIO 10) + i2s = I2S(1, + sck=Pin(9), + ws=Pin(45), + sd=Pin(10), + mode=I2S.RX, + ibuf=16000, + rate=16000, + bits=16, + format=I2S.STEREO) + + # 3. Wake up and configure the ES7210 microphone chip + mic_adc = ES7210(i2c) + if not mic_adc.init(sample_rate=16000, bit_width=16): + i2s.deinit() + return False + + print(f"Recording {duration_seconds} seconds of audio...") + + # Create reading buffer (reads 100ms chunks: 16000 samples/sec * 2 channels * 2 bytes/sample * 0.1s = 6400 bytes) + buffer = bytearray(6400) + + start_time = time.time() + total_bytes = 0 + + try: + with open(filename, 'wb') as f: + while (time.time() - start_time) < duration_seconds: + # Read raw stereo PCM data from I2S + bytes_read = i2s.readinto(buffer) + if bytes_read > 0: + f.write(buffer[:bytes_read]) + total_bytes += bytes_read + + print(f"Recording saved successfully to '{filename}' ({total_bytes} bytes).") + return True + except Exception as e: + print(f"Error during recording: {e}") + return False + finally: + # Always release the I2S peripheral resources + i2s.deinit() + print("I2S receiver deinitialized.") diff --git a/battery_util.py b/battery_util.py new file mode 100644 index 0000000..91ca8d7 --- /dev/null +++ b/battery_util.py @@ -0,0 +1,70 @@ +from machine import ADC, Pin + +class BatteryMonitor: + """Utility class for monitoring battery voltage and capacity on ESP32-S3-RLCD-4.2.""" + + def __init__(self, pin_num=4): + # Initialize ADC on GPIO 4 with 11dB attenuation + self.adc = ADC(Pin(pin_num)) + self.adc.atten(ADC.ATTN_11DB) + + def read_voltage(self): + """Reads the battery voltage in Volts using internal calibration. + + Returns: + float: Battery voltage in Volts (e.g. 4.15) or None on error. + """ + try: + # Try calibrated reading in microvolts first + uv = self.adc.read_uv() + # 3x voltage divider onboard scales 3.0V-4.2V battery to 1.0V-1.4V + voltage = (uv / 1_000_000.0) * 3.0 + return round(voltage, 3) + except AttributeError: + # Fallback if read_uv() is not supported on older MicroPython builds + try: + # Read 12-bit value (0-4095) + raw = self.adc.read() + # 3.3V reference at 11dB attenuation, 3x divider + voltage = (raw / 4095.0) * 3.3 * 3.0 + return round(voltage, 3) + except Exception as e: + print(f"Error reading battery raw ADC: {e}") + return None + except Exception as e: + print(f"Error reading battery calibrated ADC: {e}") + return None + + def read_percentage(self): + """Computes approximate battery percentage based on discharge curve. + + Assumes linear approximation between 3.0V (0%) and 4.2V (100%). + + Returns: + int: Percentage between 0 and 100. + """ + voltage = self.read_voltage() + if voltage is None: + return 0 + + # Bound typical lithium-ion limits + # 3.0V is typically empty for ESP32 systems where the LDO drops out around 3.3V + v_min = 3.0 + v_max = 4.2 + + if voltage <= v_min: + return 0 + if voltage >= v_max: + return 100 + + # Linear scaling + pct = (voltage - v_min) / (v_max - v_min) * 100.0 + return int(pct) + + def get_status_summary(self): + """Returns a string description of battery status.""" + v = self.read_voltage() + p = self.read_percentage() + if v is None: + return "Battery: Error" + return f"Battery: {v:.2f}V ({p}%)" diff --git a/ble_util.py b/ble_util.py new file mode 100644 index 0000000..b144c41 --- /dev/null +++ b/ble_util.py @@ -0,0 +1,140 @@ +import bluetooth +import struct +import time + +# BLE UUID definitions (Nordic UART Service) +_UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E") +_UART_TX = ( + bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"), + bluetooth.FLAG_NOTIFY, +) +_UART_RX = ( + bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"), + bluetooth.FLAG_WRITE | bluetooth.FLAG_WRITE_NO_RESPONSE, +) +_UART_SERVICE = (_UART_UUID, (_UART_TX, _UART_RX)) + +# BLE IRQ constants +_IRQ_CENTRAL_CONNECT = 1 +_IRQ_CENTRAL_DISCONNECT = 2 +_IRQ_GATTS_WRITE = 3 + +class BLEUART: + """A helper class to manage BLE UART (Serial Over BLE) for raw text data exchange.""" + + def __init__(self, ble=None, name="ESP32-S3-RLCD"): + self._ble = ble if ble is not None else bluetooth.BLE() + self._ble.active(True) + self._ble.irq(self._irq) + + # Register UART service + ((self._tx_handle, self._rx_handle),) = self._ble.gatts_register_services((_UART_SERVICE,)) + + self._connections = set() + self._rx_callback = None + self._name = name + + self._advertise() + + def _irq(self, event, data): + # Handle connection, disconnection, and writes + if event == _IRQ_CENTRAL_CONNECT: + conn_handle, _, _ = data + self._connections.add(conn_handle) + print(f"BLE Central connected (handle: {conn_handle})") + + elif event == _IRQ_CENTRAL_DISCONNECT: + conn_handle, _, _ = data + if conn_handle in self._connections: + self._connections.remove(conn_handle) + print(f"BLE Central disconnected (handle: {conn_handle})") + # Restart advertising + self._advertise() + + elif event == _IRQ_GATTS_WRITE: + conn_handle, value_handle = data + if value_handle == self._rx_handle: + # Read received data + data_received = self._ble.gatts_read(self._rx_handle) + if self._rx_callback: + try: + decoded = data_received.decode("utf-8").strip() + self._rx_callback(decoded) + except Exception as e: + # Fallback for binary / non-UTF-8 data + self._rx_callback(data_received) + + def _advertise(self): + # Generate advertising payload (flags + service UUID = 3 + 18 = 21 bytes) + adv_data = self._build_advertising_payload(services=[_UART_UUID], include_flags=True) + # Generate scan response payload (device name, no flags = 2 + len(name) bytes) + resp_data = self._build_advertising_payload(name=self._name, include_flags=False) + + self._ble.gap_advertise(100000, adv_data=adv_data, resp_data=resp_data) + print(f"BLE advertising started as '{self._name}'") + + def _build_advertising_payload(self, name=None, services=None, include_flags=True): + """Generates standard BLE advertising packet format.""" + payload = bytearray() + + def append(ad_type, value): + payload.append(len(value) + 1) + payload.append(ad_type) + payload.extend(value) + + # Flags: General discoverable mode, BR/EDR not supported + if include_flags: + append(0x01, struct.pack("B", 0x02 | 0x04)) + + # Name + if name: + append(0x09, name.encode("utf-8")) + + # Services UUID + if services: + for uuid in services: + # UUID bytes are in little-endian order + append(0x07, bytes(uuid)) + + return bytes(payload) + + def on_rx(self, callback): + """Register a callback function to handle incoming BLE messages. + + Args: + callback (function): Function accepting a single string argument. + """ + self._rx_callback = callback + return callback + + def write(self, data): + """Sends data (string or bytes) to all connected BLE Centrals.""" + if not self.is_connected(): + return False + + # Ensure it is bytes + if isinstance(data, str): + data = data.encode("utf-8") + + try: + # Write value to local GATT DB first + self._ble.gatts_write(self._tx_handle, data) + # Notify all connected clients + for conn_handle in self._connections: + self._ble.gatts_notify(conn_handle, self._tx_handle, data) + return True + except Exception as e: + print(f"Error sending BLE data: {e}") + return False + + def is_connected(self): + """Returns True if at least one central is connected.""" + return len(self._connections) > 0 + + def close(self): + """Disconnect and stop BLE.""" + self._ble.gap_advertise(None) # Stop advertising + for conn_handle in self._connections: + self._ble.gap_disconnect(conn_handle) + self._ble.active(False) + print("BLE closed.") diff --git a/button_util.py b/button_util.py new file mode 100644 index 0000000..e7c09b8 --- /dev/null +++ b/button_util.py @@ -0,0 +1,70 @@ +import time +from machine import Pin + +class Button: + """A debounced button handler supporting click and long press callbacks using Pin IRQ.""" + + def __init__(self, pin_num, name="Button", debounce_ms=50, long_press_ms=800): + self.pin = Pin(pin_num, Pin.IN, Pin.PULL_UP) + self.name = name + self.debounce_ms = debounce_ms + self.long_press_ms = long_press_ms + + self.last_state = 1 # Normal high (pull-up) + self.press_time = 0 + self.last_debounce_time = 0 + + self.click_callback = None + self.long_press_callback = None + + # Setup soft IRQ (default) to allow memory allocation and callbacks inside handler + self.pin.irq(handler=self._handle_irq, trigger=Pin.IRQ_FALLING | Pin.IRQ_RISING) + + def is_pressed(self): + """Returns True if the button is currently pressed (active low).""" + return self.pin.value() == 0 + + def on_click(self, callback): + """Register a callback function for a single click event.""" + self.click_callback = callback + return callback + + def on_long_press(self, callback): + """Register a callback function for a long press event.""" + self.long_press_callback = callback + return callback + + def _handle_irq(self, pin): + now = time.ticks_ms() + val = pin.value() + + # Debounce: filter out fast state jitter + if time.ticks_diff(now, self.last_debounce_time) < self.debounce_ms: + return + + if val != self.last_state: + self.last_debounce_time = now + self.last_state = val + + if val == 0: + # Button pressed + self.press_time = now + else: + # Button released + if self.press_time > 0: + duration = time.ticks_diff(now, self.press_time) + self.press_time = 0 # Reset + + if duration >= self.long_press_ms: + if self.long_press_callback: + self.long_press_callback() + else: + if self.click_callback: + self.click_callback() + +class BoardButtons: + """Convenience class containing both the BOOT (Pin 0) and KEY (Pin 18) buttons.""" + + def __init__(self): + self.boot = Button(0, "BOOT") + self.key = Button(18, "KEY") diff --git a/demo_all.py b/demo_all.py new file mode 100644 index 0000000..69d4066 --- /dev/null +++ b/demo_all.py @@ -0,0 +1,167 @@ +import time +import machine +from machine import Pin, SPI, I2C +import rlcd + +# Import our utility classes +from shtc3_util import SHTC3 +from rtc_util import PCF85063 +from battery_util import BatteryMonitor +from button_util import BoardButtons +from rgb_led_util import BoardLED + +# LED state management +led_modes = [ + ("Red (Breathing)", lambda led: led.set_color(40, 0, 0), "breath"), + ("Green (Breathing)", lambda led: led.set_color(0, 40, 0), "breath"), + ("Blue (Breathing)", lambda led: led.set_color(0, 0, 40), "breath"), + ("Cyan (Breathing)", lambda led: led.set_color(0, 30, 30), "breath"), + ("Magenta (Breathing)", lambda led: led.set_color(30, 0, 30), "breath"), + ("Rainbow Cycle", lambda led: led.set_color(30, 30, 30), "rainbow"), # Dummy base color, handled by cycle + ("LED Off", lambda led: led.off(), "off") +] +current_mode_idx = 1 # Start on Green (Breathing) + +def main(): + global current_mode_idx + print("=== Initializing ESP32-S3-RLCD-4.2 Demo ===") + + # 1. Initialize shared I2C bus and SPI display + print("Initializing I2C (SDA=13, SCL=14)...") + i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + + print("Initializing SPI and Display...") + spi = SPI(1, baudrate=20000000, polarity=0, phase=0, sck=Pin(11), mosi=Pin(12)) + display = rlcd.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) + + # Enable the speaker amplifier power control pin (just to show how, set low for now to save power/avoid hum) + print("Configuring Audio Amp control pin...") + amp_pin = Pin(46, Pin.OUT, value=0) + + # 2. Initialize Utilities + print("Initializing hardware utilities...") + sensor = SHTC3(i2c) + rtc_chip = PCF85063(i2c) + battery = BatteryMonitor() + led = BoardLED(38) + buttons = BoardButtons() + + # Sync system clock from hardware RTC on startup + rtc_chip.sync_to_system() + + # Set initial LED mode (Green breathing) + led_modes[current_mode_idx][1](led) + + # 3. Setup Button Event Callbacks + last_action_str = "System Started" + trigger_screen_update = True + + def on_key_click(): + global current_mode_idx + current_mode_idx = (current_mode_idx + 1) % len(led_modes) + mode_name = led_modes[current_mode_idx][0] + # Apply color settings + led_modes[current_mode_idx][1](led) + print(f"KEY clicked! Switched LED mode to: {mode_name}") + nonlocal last_action_str, trigger_screen_update + last_action_str = f"LED: {mode_name}" + trigger_screen_update = True + + def on_boot_click(): + print("BOOT clicked! Triggering manual refresh.") + nonlocal last_action_str, trigger_screen_update + last_action_str = "Manual Screen Refresh" + trigger_screen_update = True + + # Register callbacks + buttons.key.on_click(on_key_click) + buttons.boot.on_click(on_boot_click) + + # 4. Main execution loop + print("Starting Main Loop...") + last_screen_update = 0 + screen_update_interval_ms = 5000 # Update display every 5 seconds or on button triggers + + screenshot_saved = False + + while True: + current_time_ms = time.ticks_ms() + + # Check if we should update the screen + if trigger_screen_update or (time.ticks_diff(current_time_ms, last_screen_update) >= screen_update_interval_ms): + trigger_screen_update = False + last_screen_update = current_time_ms + + # Read sensor values + temp, hum = sensor.read_sensor() + temp_str = f"{temp} C" if temp is not None else "Error" + hum_str = f"{hum} %" if hum is not None else "Error" + + # Read RTC time + dt = rtc_chip.get_datetime() + if dt: + time_str = f"{dt[0]:04d}-{dt[1]:02d}-{dt[2]:02d} {dt[4]:02d}:{dt[5]:02d}:{dt[6]:02d}" + else: + time_str = "RTC Error" + + # Read battery info + bat_v = battery.read_voltage() + bat_p = battery.read_percentage() + bat_str = f"{bat_v:.2f}V ({bat_p}%)" if bat_v is not None else "Error" + + print(f"[{rtc_chip.get_time_string()}] T={temp_str}, H={hum_str}, Battery={bat_str}") + + # Draw display + display.clear(0) # Clear to white (standard background) + + # Title + display.text("WAVESHARE ESP32-S3-RLCD-4.2", 10, 10, 1) + display.line(10, 20, 390, 20, 1) + + # Sensor Readings block + display.text_large("ENVIRONMENT", 15, 30, scale=2, c=1) + display.text(f"Temperature : {temp_str}", 25, 55, 1) + display.text(f"Humidity : {hum_str}", 25, 70, 1) + + display.line(10, 95, 390, 95, 1) + + # Device Status block + display.text_large("DEVICE STATUS", 15, 105, scale=2, c=1) + display.text(f"RTC Time : {time_str}", 25, 130, 1) + display.text(f"Battery : {bat_str}", 25, 145, 1) + + display.line(10, 170, 390, 170, 1) + + # Interactions block + display.text_large("INTERACTIVE CONTROLS", 15, 180, scale=2, c=1) + display.text("Press [KEY] button to cycle NeoPixel LED", 25, 205, 1) + display.text("Press [BOOT] button to force display refresh", 25, 220, 1) + + # Footer / Last action + display.line(10, 245, 390, 245, 1) + display.text(f"Event: {last_action_str}", 15, 255, 1) + + # Send buffer to screen + display.show() + + # Capture the first frame as a screenshot so we can verify it + if not screenshot_saved: + try: + display.save_screenshot('screenshot_demo.pbm') + screenshot_saved = True + except Exception as e: + print(f"Failed to save screenshot_demo: {e}") + + # Update NeoPixel animation smoothly (runs every 50ms) + mode_type = led_modes[current_mode_idx][2] + if mode_type == "breath": + led.update_breathing(1.5) + elif mode_type == "rainbow": + led.update_rainbow(0.4) + else: + led.off() + + time.sleep_ms(50) + +if __name__ == "__main__": + main() diff --git a/main.py b/main.py new file mode 100644 index 0000000..efbc192 --- /dev/null +++ b/main.py @@ -0,0 +1,53 @@ +import rlcd +from machine import Pin, SPI +from wifi_util import WiFiUtil +import time + +def main(): + print("Initializing SPI and Display...") + # Setup SPI + spi = SPI(1, baudrate=20000000, polarity=0, phase=0, + sck=Pin(11), mosi=Pin(12)) + + # Initialize Display + display = rlcd.RLCD(spi, cs=Pin(40), dc=Pin(5), rst=Pin(41)) + + # 1. Show a loading state + display.clear(0) + display.text("Scanning for Wi-Fi...", 10, 10, 1) + display.show() + + # 2. Scan for networks + wifi = WiFiUtil() + networks = wifi.scan() + + # 3. Draw the results + print("Drawing networks to buffer...") + display.clear(0) # Clear to white + + # Title + display.text("Available Wi-Fi Networks:", 10, 10, 1) + display.line(10, 20, 210, 20, 1) + + # List networks + y_offset = 30 + # Limit to 20 networks so they fit on the 300px tall screen (20 * 12px = 240px + 30px offset = 270px) + for i, net in enumerate(networks[:20]): + text = f"{i+1}. {net['ssid']} ({net['rssi']}dBm)" + display.text(text, 10, y_offset, 1) + y_offset += 12 + + if not networks: + display.text("No networks found.", 10, 30, 1) + + # 4. Update Screen + print("Updating screen...") + display.show() + try: + display.save_screenshot('screenshot.pbm') + except Exception as e: + print(f"Failed to save screenshot: {e}") + print("Done!") + +if __name__ == "__main__": + main() diff --git a/rgb_led_util.py b/rgb_led_util.py new file mode 100644 index 0000000..8365bf5 --- /dev/null +++ b/rgb_led_util.py @@ -0,0 +1,61 @@ +import math +import time +from machine import Pin +import neopixel + +class BoardLED: + """Utility class to control the onboard WS2812 (NeoPixel) RGB LED on GPIO 38.""" + + def __init__(self, pin_num=38): + self.np = neopixel.NeoPixel(Pin(pin_num), 1) + self.base_color = (0, 0, 0) + self.off() + + def set_color(self, r, g, b): + """Sets the RGB LED to a specific static color. + + Values are between 0 and 255. + """ + self.base_color = (r, g, b) + self.np[0] = (r, g, b) + self.np.write() + + def off(self): + """Turns the RGB LED off.""" + self.set_color(0, 0, 0) + + def update_breathing(self, speed_factor=1.0): + """Performs one step of a non-blocking breathing animation. + + Call this regularly in the main application loop. + + Args: + speed_factor (float): Speeds up or slows down the breathing rate. + """ + if self.base_color == (0, 0, 0): + return + + t = time.ticks_ms() / 1000.0 * speed_factor + # Sine wave from 0.05 to 1.0 for breathing intensity + factor = 0.525 + 0.475 * math.sin(t * math.pi) + + br = int(self.base_color[0] * factor) + bg = int(self.base_color[1] * factor) + bb = int(self.base_color[2] * factor) + + self.np[0] = (br, bg, bb) + self.np.write() + + def update_rainbow(self, speed_factor=0.2): + """Performs one step of a non-blocking rainbow color cycle. + + Call this regularly in the main application loop. + """ + t = time.ticks_ms() / 1000.0 * speed_factor + # Use phase offsets of 120 degrees (2pi/3) for red, green, blue + r = int(127.5 * (1.0 + math.sin(t * 2.0 * math.pi))) + g = int(127.5 * (1.0 + math.sin(t * 2.0 * math.pi + 2.0 * math.pi / 3.0))) + b = int(127.5 * (1.0 + math.sin(t * 2.0 * math.pi + 4.0 * math.pi / 3.0))) + + self.np[0] = (r, g, b) + self.np.write() diff --git a/rlcd.py b/rlcd.py new file mode 100644 index 0000000..766551a --- /dev/null +++ b/rlcd.py @@ -0,0 +1,160 @@ +import time +from machine import Pin, SPI +import framebuf +import micropython + +class RLCD: + def __init__(self, spi, cs, dc, rst, width=400, height=300): + self.spi = spi + self.cs = cs + self.dc = dc + self.rst = rst + self.width = width + self.height = height + + # 1. Hardware Buffer (The weird format the screen needs) + self.hw_len = (self.width * self.height) // 8 + self.hw_buffer = bytearray(self.hw_len) + + # 2. Virtual Canvas (Standard format for drawing lines, text, bitmaps) + # We use MONO_HLSB (Standard Horizontal Byte layout) + self.canvas_buffer = bytearray(self.hw_len) + self.canvas = framebuf.FrameBuffer(self.canvas_buffer, self.width, self.height, framebuf.MONO_HLSB) + + # Init Pins + self.cs.init(self.cs.OUT, value=1) + self.dc.init(self.dc.OUT, value=0) + self.rst.init(self.rst.OUT, value=1) + + self.reset() + self.init_display() + + # --- DRAWING WRAPPERS --- + def pixel(self, x, y, c): self.canvas.pixel(x, y, c) + def line(self, x1, y1, x2, y2, c): self.canvas.line(x1, y1, x2, y2, c) + def rect(self, x, y, w, h, c): self.canvas.rect(x, y, w, h, c) + def fill_rect(self, x, y, w, h, c): self.canvas.fill_rect(x, y, w, h, c) + def text(self, msg, x, y, c=1): self.canvas.text(msg, x, y, c) + def clear(self, c=0): self.canvas.fill(c) + + # --- SCALABLE TEXT --- + def text_large(self, msg, x, y, scale=2, c=1): + char_w = 8; char_h = 8 + tmp_buf = bytearray(char_w * char_h // 8) + tmp_fb = framebuf.FrameBuffer(tmp_buf, char_w, char_h, framebuf.MONO_HLSB) + for char in msg: + tmp_fb.fill(0); tmp_fb.text(char, 0, 0, 1) + for py in range(8): + for px in range(8): + if tmp_fb.pixel(px, py): + self.canvas.fill_rect(x + (px * scale), y + (py * scale), scale, scale, c) + x += (8 * scale) + + # --- RAW BITMAPS (1:1 scale) --- + def bitmap(self, x, y, w, h, pixel_data): + img = framebuf.FrameBuffer(pixel_data, w, h, framebuf.MONO_HLSB) + self.canvas.blit(img, x, y) + + # --- PBM FILE LOADER WITH SCALING (NEW!) --- + def draw_pbm(self, filename, x, y, scale=1): + try: + with open(filename, 'rb') as f: + line1 = f.readline() + if not line1.startswith(b'P4'): print("Err: Not P4 PBM"); return + while True: + line = f.readline() + if not line.startswith(b'#'): break + dims = line.split(); w = int(dims[0]); h = int(dims[1]) + data = bytearray(f.read()) + + # Create temp buffer for source image + src_fb = framebuf.FrameBuffer(data, w, h, framebuf.MONO_HLSB) + + if scale == 1: + self.canvas.blit(src_fb, x, y) # Fast path + else: + # Slow path: iterate pixels and draw scaled rects + for sy in range(h): + for sx in range(w): + if src_fb.pixel(sx, sy): + self.canvas.fill_rect(x + (sx * scale), y + (sy * scale), scale, scale, 1) + print(f"Loaded {filename} (scale {scale})") + except OSError: + print(f"Error: Could not open {filename}") + + # --- SCREENSHOT --- + def save_screenshot(self, filename): + print(f"Saving screenshot to {filename}...") + try: + with open(filename, 'wb') as f: + f.write(b'P4\n') + f.write(f"{self.width} {self.height}\n".encode()) + f.write(self.canvas_buffer) + print("Saved!") + except Exception as e: + print(f"Error saving screenshot: {e}") + + # --- HARDWARE LOGIC --- + def reset(self): + self.rst(1); time.sleep_ms(50); self.rst(0); time.sleep_ms(20); self.rst(1); time.sleep_ms(50) + + def write_cmd(self, cmd): + self.cs(0); self.dc(0); self.spi.write(bytearray([cmd])); self.cs(1) + + def write_data(self, data): + self.cs(0); self.dc(1) + if isinstance(data, int): self.spi.write(bytearray([data])) + elif isinstance(data, list): self.spi.write(bytearray(data)) + else: self.spi.write(data) + self.cs(1) + + def init_display(self): + self.write_cmd(0xD6); self.write_data(0x17); self.write_data(0x02) + self.write_cmd(0xD1); self.write_data(0x01) + self.write_cmd(0xC0); self.write_data(0x11); self.write_data(0x04) + self.write_cmd(0xC1); self.write_data([0x41, 0x41, 0x41, 0x41]) + self.write_cmd(0xC2); self.write_data([0x19, 0x19, 0x19, 0x19]) + self.write_cmd(0xC4); self.write_data([0x41, 0x41, 0x41, 0x41]) + self.write_cmd(0xC5); self.write_data([0x19, 19, 0x19, 0x19]) + self.write_cmd(0xD8); self.write_data(0xA6); self.write_data(0xE9) + self.write_cmd(0xB2); self.write_data(0x05) + self.write_cmd(0xB3); self.write_data([0xE5, 0xF6, 0x05, 0x46, 0x77, 0x77, 0x77, 0x77, 0x76, 0x45]) + self.write_cmd(0xB4); self.write_data([0x05, 0x46, 0x77, 0x77, 0x77, 0x77, 0x76, 0x45]) + self.write_cmd(0x62); self.write_data([0x32, 0x03, 0x1F]) + self.write_cmd(0xB7); self.write_data(0x13) + self.write_cmd(0xB0); self.write_data(0x64) + self.write_cmd(0x11); time.sleep_ms(200) + self.write_cmd(0xC9); self.write_data(0x00) + self.write_cmd(0x36); self.write_data(0x48) + self.write_cmd(0x3A); self.write_data(0x11) + self.write_cmd(0xB9); self.write_data(0x20) + self.write_cmd(0xB8); self.write_data(0x29) + self.write_cmd(0x20) # Inversion OFF (White Background) + self.write_cmd(0x2A); self.write_data([0x12, 0x2A]) + self.write_cmd(0x2B); self.write_data([0x00, 0xC7]) + self.write_cmd(0x35); self.write_data(0x00) + self.write_cmd(0xD0); self.write_data(0xFF) + self.write_cmd(0x38); self.write_cmd(0x29) + + # --- THE HEAVY LIFTER (Optimized) --- + @micropython.native + def show(self): + HEIGHT = 300; WIDTH = 400 + for i in range(len(self.hw_buffer)): self.hw_buffer[i] = 0 + for y in range(HEIGHT): + for x in range(WIDTH): + if self.canvas.pixel(x, y): + inv_y = HEIGHT - 1 - y + byte_x = x // 2 + block_y = inv_y // 4 + index = byte_x * 75 + block_y + local_x = x % 2 + local_y = inv_y % 4 + bit = 7 - (local_y * 2 + local_x) + self.hw_buffer[index] |= (1 << bit) + self.write_cmd(0x2A); self.write_data(0x12); self.write_data(0x2A) + self.write_cmd(0x2B); self.write_data(0x00); self.write_data(0xC7) + self.write_cmd(0x2C) + self.cs(0); self.dc(1) + self.spi.write(self.hw_buffer) + self.cs(1) \ No newline at end of file diff --git a/rtc_util.py b/rtc_util.py new file mode 100644 index 0000000..9e98502 --- /dev/null +++ b/rtc_util.py @@ -0,0 +1,122 @@ +import time +from machine import I2C, Pin, RTC + +class PCF85063: + """Utility class for the PCF85063 Real-Time Clock (RTC) chip.""" + ADDR = 0x51 + + # Starting register for timekeeping data + TIME_REG_START = 0x04 + + def __init__(self, i2c=None): + if i2c is None: + # Default to ESP32-S3-RLCD-4.2 onboard I2C pins + self.i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + else: + self.i2c = i2c + + # Initialize RTC if needed (ensure oscillator is running) + self._init_rtc() + + def _init_rtc(self): + try: + # Control register 1 (0x00) should be 0 to ensure oscillator is not stopped (bit 5 = STOP) + ctrl1 = self.i2c.readfrom_mem(self.ADDR, 0x00, 1)[0] + if ctrl1 & 0x20: # Oscillator is stopped + print("RTC oscillator was stopped. Starting oscillator...") + self.i2c.writeto_mem(self.ADDR, 0x00, b'\x00') + except Exception as e: + print(f"Error initializing PCF85063: {e}") + + def _dec2bcd(self, val): + return (val // 10 << 4) | (val % 10) + + def _bcd2dec(self, val): + return ((val >> 4) * 10) + (val & 0x0F) + + def get_datetime(self): + """Reads current time from hardware RTC. + + Returns: + tuple: (year, month, day, weekday, hour, minute, second) or None on error. + Note: year is 4-digit (e.g. 2026). + """ + try: + # Read 7 registers: Seconds, Minutes, Hours, Days, Weekdays, Months, Years + data = self.i2c.readfrom_mem(self.ADDR, self.TIME_REG_START, 7) + + second = self._bcd2dec(data[0] & 0x7F) # Bit 7 is integrity flag OS + minute = self._bcd2dec(data[1] & 0x7F) + hour = self._bcd2dec(data[2] & 0x3F) # 24-hour mode + day = self._bcd2dec(data[3] & 0x3F) + weekday = data[4] & 0x07 # 0-6 + month = self._bcd2dec(data[5] & 0x1F) + year = 2000 + self._bcd2dec(data[6]) # 0-99 mapping to 2000-2099 + + return (year, month, day, weekday, hour, minute, second) + except Exception as e: + print(f"Error reading PCF85063 RTC: {e}") + return None + + def set_datetime(self, dt): + """Sets the hardware RTC time. + + Args: + dt (tuple): (year, month, day, weekday, hour, minute, second) + Note: year can be 4-digit (e.g. 2026) or 2-digit (e.g. 26). + """ + try: + year, month, day, weekday, hour, minute, second = dt + + # Format to registers + reg_year = year % 100 + + data = bytearray(7) + data[0] = self._dec2bcd(second) & 0x7F # Clear OS bit (re-starts oscillator integrity check) + data[1] = self._dec2bcd(minute) + data[2] = self._dec2bcd(hour) + data[3] = self._dec2bcd(day) + data[4] = weekday & 0x07 + data[5] = self._dec2bcd(month) + data[6] = self._dec2bcd(reg_year) + + self.i2c.writeto_mem(self.ADDR, self.TIME_REG_START, data) + return True + except Exception as e: + print(f"Error setting PCF85063 RTC: {e}") + return False + + def sync_to_system(self): + """Synchronizes the ESP32-S3 internal system time from this hardware RTC.""" + dt = self.get_datetime() + if dt: + year, month, day, weekday, hour, minute, second = dt + # machine.RTC.datetime requires 8-tuple: (year, month, day, weekday, hour, minute, second, subseconds) + sys_rtc = RTC() + sys_rtc.datetime((year, month, day, weekday, hour, minute, second, 0)) + print(f"System clock synced to RTC: {year:04d}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}:{second:02d}") + return True + return False + + def sync_from_system(self): + """Synchronizes the hardware RTC time from the ESP32-S3 system clock.""" + try: + # Get current system time (localtime tuple: year, month, mday, hour, minute, second, weekday, yearday) + t = time.localtime() + # mapping time.localtime() weekday to PCF85063: + # Python weekday (0=Monday...6=Sunday) + dt = (t[0], t[1], t[2], t[6], t[3], t[4], t[5]) + success = self.set_datetime(dt) + if success: + print(f"RTC synced from System: {t[0]:04d}-{t[1]:02d}-{t[2]:02d} {t[3]:02d}:{t[4]:02d}:{t[5]:02d}") + return success + except Exception as e: + print(f"Error syncing RTC from system: {e}") + return False + + def get_time_string(self): + """Returns current time as a formatted string: YYYY-MM-DD HH:MM:SS""" + dt = self.get_datetime() + if dt: + return f"{dt[0]:04d}-{dt[1]:02d}-{dt[2]:02d} {dt[4]:02d}:{dt[5]:02d}:{dt[6]:02d}" + return "0000-00-00 00:00:00" diff --git a/sd_util.py b/sd_util.py new file mode 100644 index 0000000..c486a05 --- /dev/null +++ b/sd_util.py @@ -0,0 +1,97 @@ +import os +import machine +from machine import Pin, SDCard + +class SDCardManager: + """Utility class to mount and manage the onboard microSD card slot. + + Uses SDMMC 1-bit mode on Pins: sck=2, cmd=1, data0=3. + """ + + def __init__(self, mount_point='/sd'): + self.mount_point = mount_point + self.sd = None + self.mounted = False + + def mount(self): + """Initializes the SD card and mounts it to the filesystem. + + Returns: + bool: True if mounted successfully, False otherwise. + """ + if self.mounted: + print(f"SD card already mounted at {self.mount_point}") + return True + + try: + print("Initializing SDCard (SDMMC 1-bit mode: sck=2, cmd=1, data0=3)...") + # slot=1 uses the SDMMC peripheral on the ESP32-S3 + self.sd = SDCard(slot=1, width=1, sck=2, cmd=1, data=(3,)) + + print(f"Mounting SD card to {self.mount_point}...") + os.mount(self.sd, self.mount_point) + self.mounted = True + print("SD card mounted successfully!") + return True + except Exception as e: + print(f"Failed to mount SD card: {e}") + self.sd = None + self.mounted = False + return False + + def unmount(self): + """Unmounts the SD card from the filesystem.""" + if not self.mounted: + return True + + try: + print(f"Unmounting SD card from {self.mount_point}...") + os.umount(self.mount_point) + self.mounted = False + self.sd = None + print("SD card unmounted.") + return True + except Exception as e: + print(f"Failed to unmount SD card: {e}") + return False + + def is_mounted(self): + """Returns True if the card is currently mounted.""" + return self.mounted + + def list_files(self): + """Lists files on the SD card if mounted. + + Returns: + list: List of filenames or None if not mounted. + """ + if not self.mounted: + print("SD card is not mounted.") + return None + try: + return os.listdir(self.mount_point) + except Exception as e: + print(f"Error listing SD card files: {e}") + return None + + def get_info(self): + """Returns storage information of the SD card. + + Returns: + dict: {total_bytes, free_bytes} or None on error. + """ + if not self.mounted: + return None + try: + stat = os.statvfs(self.mount_point) + block_size = stat[0] + total_blocks = stat[2] + free_blocks = stat[3] + + return { + "total_bytes": total_blocks * block_size, + "free_bytes": free_blocks * block_size + } + except Exception as e: + print(f"Error getting SD card info: {e}") + return None diff --git a/shtc3_util.py b/shtc3_util.py new file mode 100644 index 0000000..799ca34 --- /dev/null +++ b/shtc3_util.py @@ -0,0 +1,85 @@ +import time +from machine import I2C, Pin + +class SHTC3: + """Utility class for Sensirion SHTC3 Temperature and Humidity Sensor.""" + ADDR = 0x70 + + # Commands + WAKE = b'\x35\x17' + SLEEP = b'\xB0\x98' + MEASURE = b'\x78\x66' # High precision, T first, clock stretching disabled + + def __init__(self, i2c=None): + if i2c is None: + # Default to ESP32-S3-RLCD-4.2 onboard I2C pins + self.i2c = I2C(0, sda=Pin(13), scl=Pin(14)) + else: + self.i2c = i2c + + def _crc8(self, data): + """Sensirion CRC-8 checksum calculation (polynom 0x31, init 0xFF).""" + crc = 0xFF + for byte in data: + crc ^= byte + for _ in range(8): + if crc & 0x80: + crc = (crc << 1) ^ 0x31 + else: + crc <<= 1 + crc &= 0xFF + return crc + + def read_sensor(self): + """Wakes up the sensor, reads temp and humidity, validates CRC, and sleeps. + + Returns: + tuple: (temperature_c, relative_humidity_pct) or (None, None) on error. + """ + try: + # 1. Wakeup + self.i2c.writeto(self.ADDR, self.WAKE) + time.sleep_ms(1) + + # 2. Trigger Measurement + self.i2c.writeto(self.ADDR, self.MEASURE) + time.sleep_ms(15) # High precision measurement takes up to 12.1ms + + # 3. Read 6 bytes of data + # bytes 0, 1: Temp, byte 2: Temp CRC + # bytes 3, 4: Hum, byte 5: Hum CRC + data = self.i2c.readfrom(self.ADDR, 6) + + # 4. Enter low-power sleep mode + self.i2c.writeto(self.ADDR, self.SLEEP) + + # Verify CRC + t_data = data[0:2] + t_crc = data[2] + h_data = data[3:5] + h_crc = data[5] + + if self._crc8(t_data) != t_crc: + print("SHTC3 Temp CRC error") + return None, None + if self._crc8(h_data) != h_crc: + print("SHTC3 Hum CRC error") + return None, None + + raw_t = (data[0] << 8) | data[1] + raw_h = (data[3] << 8) | data[4] + + # Formula from datasheet + temp = -45.0 + 175.0 * (raw_t / 65536.0) + hum = 100.0 * (raw_h / 65536.0) + + return round(temp, 2), round(hum, 2) + + except Exception as e: + print(f"Error reading SHTC3 sensor: {e}") + # Try to send sleep command anyway to save power + try: + self.i2c.writeto(self.ADDR, self.SLEEP) + except: + pass + return None, None diff --git a/vector.py b/vector.py new file mode 100644 index 0000000..84bf7f8 --- /dev/null +++ b/vector.py @@ -0,0 +1,102 @@ +# Simple Vector (Stroke) Font Engine with THICKNESS support +# Format: [x1,y1, x2,y2, x3,y3...] (-1 = Pen Lift) + +FONT = { + 'A': [0,10, 5,0, 10,10, -1,-1, 2,7, 8,7], + 'B': [0,10, 0,0, 7,0, 7,5, 0,5, -1,-1, 0,5, 8,5, 8,10, 0,10], + 'C': [10,2, 8,0, 2,0, 0,2, 0,8, 2,10, 8,10, 10,8], + 'D': [0,10, 0,0, 6,0, 9,3, 9,7, 6,10, 0,10], + 'E': [10,0, 0,0, 0,10, 10,10, -1,-1, 0,5, 8,5], + 'F': [0,10, 0,0, 9,0, -1,-1, 0,5, 8,5], + 'G': [10,2, 8,0, 2,0, 0,2, 0,8, 2,10, 8,10, 10,8, 10,5, 6,5], + 'H': [0,0, 0,10, -1,-1, 10,0, 10,10, -1,-1, 0,5, 10,5], + 'I': [5,0, 5,10, -1,-1, 2,0, 8,0, -1,-1, 2,10, 8,10], + 'J': [8,0, 8,8, 6,10, 2,10, 0,8], + 'K': [0,0, 0,10, -1,-1, 9,0, 0,5, 9,10], + 'L': [0,0, 0,10, 9,10], + 'M': [0,10, 0,0, 5,5, 10,0, 10,10], + 'N': [0,10, 0,0, 10,10, 10,0], + 'O': [5,0, 10,2, 10,8, 5,10, 0,8, 0,2, 5,0], + 'P': [0,10, 0,0, 8,0, 8,5, 0,5], + 'Q': [5,0, 10,2, 10,8, 5,10, 0,8, 0,2, 5,0, -1,-1, 6,6, 10,10], + 'R': [0,10, 0,0, 8,0, 8,5, 0,5, -1,-1, 4,5, 9,10], + 'S': [9,2, 5,0, 1,2, 1,4, 5,5, 9,6, 9,8, 5,10, 1,8], + 'T': [5,0, 5,10, -1,-1, 0,0, 10,0], + 'U': [0,0, 0,8, 2,10, 8,10, 10,8, 10,0], + 'V': [0,0, 5,10, 10,0], + 'W': [0,0, 2,10, 5,6, 8,10, 10,0], + 'X': [0,0, 10,10, -1,-1, 10,0, 0,10], + 'Y': [0,0, 5,5, 10,0, -1,-1, 5,5, 5,10], + 'Z': [0,0, 10,0, 0,10, 10,10], + '0': [5,0, 10,2, 10,8, 5,10, 0,8, 0,2, 5,0, -1,-1, 0,10, 10,0], + '1': [2,2, 5,0, 5,10, -1,-1, 2,10, 8,10], + '2': [0,2, 5,0, 10,2, 10,5, 0,10, 10,10], + '3': [0,2, 5,0, 10,2, 10,4, 6,5, 10,6, 10,8, 5,10, 0,8], + '4': [7,10, 7,0, -1,-1, 0,0, 0,5, 10,5], + '5': [9,0, 1,0, 1,4, 5,3, 9,4, 9,8, 5,10, 1,8], + '6': [9,0, 1,4, 1,8, 5,10, 9,8, 9,5, 1,5], + '7': [0,0, 10,0, 4,10], + '8': [5,0, 10,2, 10,4, 5,5, 10,6, 10,8, 5,10, 0,8, 0,6, 5,5, 0,4, 0,2, 5,0], + '9': [1,10, 9,6, 9,2, 5,0, 1,2, 1,5, 9,5], + '.': [4,10, 6,10, 6,8, 4,8, 4,10], + ':': [4,2, 4,4, 5,4, 5,2, 6,2, 6,4, -1,-1, 4,8, 4,10, 5,10, 5,8, 6,8, 6,10], + '-': [1,5, 9,5], + ' ': [] +} + +def draw(display, text, x, y, scale=1, c=1, thickness=1): + start_x = x + char_width = 12 * scale + + for char in text: + char = char.upper() + if char == '\n': + y += 15 * scale + x = start_x + continue + if char == ' ': + x += char_width + continue + + if char in FONT: + strokes = FONT[char] + if len(strokes) > 1: + curr_x = strokes[0] + curr_y = strokes[1] + i = 2 + while i < len(strokes): + next_x = strokes[i] + next_y = strokes[i+1] + + if next_x == -1: # Pen Lift + i += 2 + if i < len(strokes): + curr_x = strokes[i] + curr_y = strokes[i+1] + i += 2 + continue + + # Calculate scaled coordinates + x1 = int(x + curr_x * scale) + y1 = int(y + curr_y * scale) + x2 = int(x + next_x * scale) + y2 = int(y + next_y * scale) + + # Draw Main Line + display.line(x1, y1, x2, y2, c) + + # Draw Thick Lines (Offset by 1px) + if thickness > 1: + # Draw parallel line X+1 + display.line(x1+1, y1, x2+1, y2, c) + if thickness > 2: + # Draw parallel line Y+1 + display.line(x1, y1+1, x2, y2+1, c) + if thickness > 3: + # Draw parallel line X+1, Y+1 + display.line(x1+1, y1+1, x2+1, y2+1, c) + + curr_x = next_x + curr_y = next_y + i += 2 + x += char_width \ No newline at end of file diff --git a/wifi_util.py b/wifi_util.py new file mode 100644 index 0000000..60bd2c4 --- /dev/null +++ b/wifi_util.py @@ -0,0 +1,28 @@ +import network + +class WiFiUtil: + def __init__(self): + self.wlan = network.WLAN(network.STA_IF) + self.wlan.active(True) + + def scan(self): + """Scans for networks and returns a list of dictionaries with ssid and rssi.""" + print("Scanning for Wi-Fi networks...") + networks = self.wlan.scan() + result = [] + for net in networks: + # network tuple: (ssid, bssid, channel, RSSI, authmode, hidden) + try: + ssid = net[0].decode('utf-8') + except Exception: + try: + ssid = ''.join(chr(b) if 32 <= b < 127 else '?' for b in net[0]) + except Exception: + ssid = "" + rssi = net[3] + if ssid: # Filter out hidden/empty SSIDs + result.append({"ssid": ssid, "rssi": rssi}) + + # Sort by signal strength (RSSI) descending + result.sort(key=lambda x: x['rssi'], reverse=True) + return result