Initial commit: ESP32-S3-RLCD-4.2 hardware utility classes and demo
This commit is contained in:
Adolfo Reyna
+12
@@ -0,0 +1,12 @@
|
||||
# Temporary generated media and screenshots
|
||||
*.pbm
|
||||
*.png
|
||||
*.pcm
|
||||
*.wav
|
||||
|
||||
# Python temporary files
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
# OS generated files
|
||||
.DS_Store
|
||||
+128
@@ -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.")
|
||||
@@ -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}%)"
|
||||
+140
@@ -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.")
|
||||
@@ -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")
|
||||
+167
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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)
|
||||
+122
@@ -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"
|
||||
+97
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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 = "<Unknown 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
|
||||
Reference in New Issue
Block a user