adolforeyna/mcp_screen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reorganize drivers and utility modules into lib/ subdirectory and update server and upload scripts

Browse Source
This commit is contained in:
Adolfo Reyna
2026-06-17 22:22:02 -04:00
parent 2813c11104
commit 75475723fa
16 changed files with 31 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/audio_util.py
+406
View File
@@ -0,0 +1,406 @@
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=16, SCL=15)
i2c = I2C(0, sda=Pin(16), scl=Pin(15))
# 1a. Setup Master Clock (MCLK) on GPIO 4 using PWM (needed for ES8311 ADC)
mclk_pin = Pin(4, Pin.OUT)
mclk_pwm = machine.PWM(mclk_pin)
mclk_pwm.freq(6144000)
mclk_pwm.duty_u16(32768)
# 2. Configure I2S Receiver
# Pins: sck=BCLK (GPIO 5), ws=WS/LRCK (GPIO 7), sd=DIN (GPIO 6)
i2s = I2S(1,
sck=Pin(5),
ws=Pin(7),
sd=Pin(6),
mode=I2S.RX,
ibuf=16000,
rate=16000,
bits=16,
format=I2S.STEREO)
# 3. Wake up and configure the ES8311 codec chip
mic_adc = ES8311(i2c)
if not mic_adc.init(sample_rate=16000):
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()
mclk_pwm.deinit()
print("I2S receiver and MCLK deinitialized.")
class ES8311:
"""MicroPython driver for the ES8311 Audio Codec (Speaker DAC).
Controls the ES8311 chip over I2C to configure clocks, audio format, and volume.
"""
ADDR = 0x18
def __init__(self, i2c):
self.i2c = i2c
def init(self, sample_rate=16000):
"""Initializes the ES8311 registers for audio playback.
Args:
sample_rate (int): Audio sample rate (typically 16000).
Returns:
bool: True if initialization was successful, False otherwise.
"""
print("Initializing ES8311 Speaker DAC...")
try:
# 1. Reset the chip
self._write(0x00, 0x1F)
time.sleep_ms(10)
self._write(0x00, 0x00)
time.sleep_ms(10)
# Clock Configuration (16kHz sample rate, MCLK=6.144MHz)
self._write(0x01, 0x3F) # Enable all clocks, use MCLK pin
self._write(0x02, 0x48) # pre_div=3, pre_mult=1
self._write(0x03, 0x10) # fs_mode=0, adc_osr=16
self._write(0x04, 0x10) # dac_osr=16
self._write(0x05, 0x00) # adc_div=1, dac_div=1
self._write(0x06, 0x03) # bclk_div=4 (4-1=3)
self._write(0x07, 0x00) # lrck_h=0
self._write(0x08, 0xFF) # lrck_l=255
# Audio Format Configuration (I2S standard format, 16-bit)
self._write(0x09, 0x0C) # SDP in: 16-bit I2S
self._write(0x0A, 0x0C) # SDP out: 16-bit I2S
# System / DAC Power Up
self._write(0x0D, 0x01) # Power up analog circuitry
self._write(0x0E, 0x02) # Enable analog PGA, enable ADC modulator
self._write(0x12, 0x00) # Power up DAC
self._write(0x13, 0x10) # Enable output to HP drive (speaker/hp output)
self._write(0x1C, 0x6A) # ADC Equalizer bypass
self._write(0x37, 0x08) # Bypass DAC equalizer
# Set Volume (0xBF = 0dB)
self._write(0x32, 0xBF)
# Unmute DAC
self._write(0x31, 0x00)
# Power On
self._write(0x00, 0x80)
print("ES8311 initialization complete.")
return True
except Exception as e:
print(f"Failed to initialize ES8311: {e}")
return False
def set_volume(self, val):
"""Sets DAC digital volume (0-100 scale)."""
# Volume register 0x32 accepts values from 0 (mute) to 255 (+0dB / max volume).
reg_val = int((val / 100.0) * 255.0)
reg_val = max(0, min(255, reg_val))
try:
self._write(0x32, reg_val)
except Exception as e:
print(f"Failed to set volume: {e}")
def _write(self, reg, val):
self.i2c.writeto_mem(self.ADDR, reg, bytes([val]))
def play_tone(frequency=440, duration_ms=1000, volume=50):
"""Plays a pure sine wave tone on the board speaker.
Args:
frequency (int): Tone frequency in Hz (e.g. 440 for A4).
duration_ms (int): Tone duration in milliseconds.
volume (int): Volume level from 0 to 100.
"""
import math
import struct
print(f"Playing tone: {frequency}Hz for {duration_ms}ms (vol={volume})...")
# 1. Setup Master Clock (MCLK) on GPIO 4 using PWM
mclk_pin = Pin(4, Pin.OUT)
mclk_pwm = machine.PWM(mclk_pin)
mclk_pwm.freq(6144000)
mclk_pwm.duty_u16(32768)
# 2. Start I2C Control Bus (SDA=16, SCL=15)
i2c = I2C(0, sda=Pin(16), scl=Pin(15))
# 3. Initialize the ES8311 DAC
dac = ES8311(i2c)
if not dac.init(sample_rate=16000):
mclk_pwm.deinit()
return False
dac.set_volume(volume)
# 4. Configure I2S TX
# Pins: sck=BCLK (GPIO 5), ws=WS/LRCK (GPIO 7), sd=DOUT (GPIO 8)
i2s = I2S(1,
sck=Pin(5),
ws=Pin(7),
sd=Pin(8),
mode=I2S.TX,
ibuf=8000,
rate=16000,
bits=16,
format=I2S.STEREO)
# 5. Enable Speaker Amplifier (GPIO 1, Active Low)
amp_pin = Pin(1, Pin.OUT, value=0)
# 6. Generate sine wave cycle
# Approximate frequency to make integer number of samples per cycle
# (avoiding phase clicking)
N = int(16000 / frequency)
N = max(4, N) # prevent division by zero or extremely high frequencies
volume_scale = int((volume / 100.0) * 32767)
cycle_data = bytearray()
for i in range(N):
val = int(volume_scale * math.sin(2 * math.pi * i / N))
cycle_data.extend(struct.pack("<hh", val, val)) # Stereo (L/R)
cycle_bytes = bytes(cycle_data)
# Write to I2S in chunks
total_samples = int(16000 * duration_ms / 1000)
total_cycles = int(total_samples / N)
written_cycles = 0
while written_cycles < total_cycles:
cycles_to_write = min(total_cycles - written_cycles, 100)
i2s.write(cycle_bytes * cycles_to_write)
written_cycles += cycles_to_write
# 7. Clean up
time.sleep_ms(100) # Let the remaining buffer play out
amp_pin.value(1) # Disable amp
i2s.deinit()
mclk_pwm.deinit()
print("Tone playback complete.")
return True
def play_wav(filename, volume=50):
"""Plays a standard WAV audio file on the board speaker.
Standard format: 16-bit PCM, 16kHz sample rate (recommended).
"""
import struct
print(f"Playing WAV: {filename} (vol={volume})...")
try:
f = open(filename, 'rb')
except OSError:
print(f"Error: Cannot open WAV file '{filename}'")
return False
try:
# 1. Parse WAV header chunk by chunk
riff_header = f.read(12)
if len(riff_header) < 12 or riff_header[0:4] != b'RIFF' or riff_header[8:12] != b'WAVE':
print("Error: Invalid WAV file format")
f.close()
return False
channels = 1
sample_rate = 16000
bits = 16
while True:
chunk_header = f.read(8)
if len(chunk_header) < 8:
break
chunk_id, chunk_size = struct.unpack('<4sI', chunk_header)
if chunk_id == b'fmt ':
fmt_data = f.read(chunk_size)
if len(fmt_data) >= 16:
audio_format, channels, sample_rate, byte_rate, block_align, bits = struct.unpack('<HHIIHH', fmt_data[:16])
if audio_format != 1:
print(f"Warning: Non-PCM audio format ({audio_format})")
else:
print("Error: fmt chunk too small")
f.close()
return False
elif chunk_id == b'data':
# Audio data begins immediately after this chunk size
break
else:
# Skip unknown chunk (align to even byte)
skip_bytes = (chunk_size + 1) & ~1
f.seek(skip_bytes, 1)
print(f"WAV Info: {sample_rate}Hz, {bits} bits, {'Mono' if channels == 1 else 'Stereo'}")
# 2. Setup Master Clock (MCLK) on GPIO 4 using PWM
mclk_pin = Pin(4, Pin.OUT)
mclk_pwm = machine.PWM(mclk_pin)
mclk_pwm.freq(6144000)
mclk_pwm.duty_u16(32768)
# 3. Start I2C Control Bus (SDA=16, SCL=15)
i2c = I2C(0, sda=Pin(16), scl=Pin(15))
# 4. Initialize the ES8311 DAC
dac = ES8311(i2c)
if not dac.init(sample_rate=16000):
mclk_pwm.deinit()
f.close()
return False
dac.set_volume(volume)
# 5. Configure I2S TX
# Pins: sck=BCLK (GPIO 5), ws=WS/LRCK (GPIO 7), sd=DOUT (GPIO 8)
i2s_format = I2S.MONO if channels == 1 else I2S.STEREO
i2s = I2S(1,
sck=Pin(5),
ws=Pin(7),
sd=Pin(8),
mode=I2S.TX,
ibuf=4096,
rate=sample_rate,
bits=bits,
format=i2s_format)
# 6. Enable Speaker Amplifier (GPIO 1, Active Low)
amp_pin = Pin(1, Pin.OUT, value=0)
# 7. Read and stream chunks to I2S
buf = bytearray(2048)
while True:
bytes_read = f.readinto(buf)
if bytes_read == 0:
break
i2s.write(buf[:bytes_read])
# 8. Clean up
time.sleep_ms(100) # Let the remaining buffer play out
amp_pin.value(1) # Disable amp
i2s.deinit()
mclk_pwm.deinit()
f.close()
print("WAV playback complete.")
return True
except Exception as e:
print(f"Error during WAV playback: {e}")
try:
f.close()
except:
pass
return False
lib/battery_util.py
+71
View File
@@ -0,0 +1,71 @@
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=9):
import sys
self.adc = ADC(Pin(pin_num))
if sys.platform == 'esp32':
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()
# 2x voltage divider onboard scales 3.0V-4.2V battery to 1.5V-2.1V
voltage = (uv / 1_000_000.0) * 2.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, 2x divider
voltage = (raw / 4095.0) * 3.3 * 2.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}%)"
lib/ble_util.py
+194
View File
@@ -0,0 +1,194 @@
try:
import bluetooth
has_ble = True
except ImportError:
has_ble = False
import struct
import time
if not has_ble:
class BLEUART:
def __init__(self, ble=None, name="ESP32-S3-RLCD"):
print("BLE not supported on this platform.")
def on_rx(self, callback): pass
def write(self, data): return False
def is_connected(self): return False
def scan(self, duration_ms=3000): return {}
def close(self): pass
else:
# 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 _ActiveBLEUART:
"""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._scan_results = {}
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)
elif event == 5: # _IRQ_GAP_SCAN_RESULT
addr_type, addr, adv_type, rssi, adv_data = data
# Convert address to string hex format
mac = ':'.join(f'{b:02x}' for b in addr)
name = ""
try:
# Parse advertising payload to extract complete (0x09) or short (0x08) name
i = 0
while i < len(adv_data):
length = adv_data[i]
if length == 0 or i + length + 1 > len(adv_data):
break
ad_type = adv_data[i+1]
if ad_type in (0x08, 0x09):
name = adv_data[i+2 : i+1+length].decode('utf-8')
break
i += length + 1
except:
pass
self._scan_results[mac] = {"rssi": rssi, "name": name}
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 scan(self, duration_ms=3000):
"""Scans for nearby BLE devices and returns a dictionary of results.
Returns:
dict: { 'mac_address': { 'rssi': -70, 'name': 'DeviceName' } }
"""
self._scan_results = {}
# Start passive scan (active=True requests scan responses to get names)
# Scan parameters: duration, interval (20ms), window (10ms), active (True)
self._ble.gap_scan(duration_ms, 20000, 10000, True)
time.sleep_ms(duration_ms + 100)
self._ble.gap_scan(None) # Ensure scan stops
return self._scan_results
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.")
if has_ble:
BLEUART = _ActiveBLEUART
lib/button_util.py
+70
View File
@@ -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")
lib/download_util.py
+68
View File
@@ -0,0 +1,68 @@
import os
import urequests
from sd_util import SDCardManager
def download_file(url, dest_filename, use_sd=True):
"""Downloads a file from a URL over the network in memory-efficient chunks.
If use_sd is True, it automatically mounts the microSD card and saves to /sd/dest_filename.
Otherwise, it saves to the local flash storage.
Returns:
str: The full path to the downloaded file on success, None on failure.
"""
dest_path = dest_filename
sd_manager = None
if use_sd:
sd_manager = SDCardManager(mount_point='/sd')
if not sd_manager.mount():
print("Download Error: Could not mount SD card.")
return None
dest_path = f"/sd/{dest_filename}"
print(f"Starting download from: {url} -> {dest_path}")
try:
res = urequests.get(url, stream=True)
except Exception as e:
print(f"HTTP Connection failed: {e}")
return None
if res.status_code != 200:
print(f"HTTP Error: Received status code {res.status_code}")
res.close()
return None
try:
# Read from socket stream in chunks to avoid OutOfMemory errors
chunk_size = 4096
total_downloaded = 0
with open(dest_path, 'wb') as f:
while True:
# Read chunk from the raw socket connection stream
chunk = res.raw.read(chunk_size)
if not chunk:
break
f.write(chunk)
total_downloaded += len(chunk)
# Dynamic logging
if total_downloaded % (chunk_size * 25) == 0:
print(f"Downloaded {total_downloaded // 1024} KB...")
print(f"Download complete! Saved {total_downloaded} bytes to '{dest_path}'.")
return dest_path
except Exception as e:
print(f"Error writing to file: {e}")
# Clean up partial file on failure
try:
os.remove(dest_path)
except:
pass
return None
finally:
res.close()
lib/ft6336u.py
+133
View File
@@ -0,0 +1,133 @@
import time
from machine import Pin, I2C
class FT6336U:
ADDR = 0x38
# Registers
REG_DEV_MODE = 0x00
REG_TD_STATUS = 0x02
REG_P1_XH = 0x03
REG_P1_XL = 0x04
REG_P1_YH = 0x05
REG_P1_YL = 0x06
REG_CTRL = 0x86
REG_CHIPID = 0xA3
REG_G_MODE = 0xA4
# Modes
CTRL_KEEP_ACTIVE = 0x00
G_MODE_TRIGGER = 0x01
def __init__(self, i2c, rst_pin=28, int_pin=25, width=480, height=320, swap_xy=True, invert_x=True, invert_y=False):
self.i2c = i2c
self.rst = Pin(rst_pin, Pin.OUT)
self.int = Pin(int_pin, Pin.IN, Pin.PULL_UP)
self.width = width
self.height = height
self.swap_xy = swap_xy
self.invert_x = invert_x
self.invert_y = invert_y
self.initialized = False
self.reset()
self.init_chip()
def reset(self):
self.rst(0)
time.sleep_ms(10)
self.rst(1)
time.sleep_ms(300) # Wait for chip to wake up
def read_reg(self, reg, n=1):
try:
return self.i2c.readfrom_mem(self.ADDR, reg, n)
except Exception as e:
print(f"I2C read failed at reg 0x{reg:02X}: {e}")
return None
def write_reg(self, reg, val):
try:
self.i2c.writeto_mem(self.ADDR, reg, bytearray([val]))
return True
except Exception as e:
print(f"I2C write failed at reg 0x{reg:02X}: {e}")
return False
def init_chip(self):
# 1. Read Chip ID
chip_id = self.read_reg(self.REG_CHIPID)
if chip_id is None or chip_id[0] != 0x64:
# Retry once
time.sleep_ms(100)
chip_id = self.read_reg(self.REG_CHIPID)
if chip_id is None or chip_id[0] != 0x64:
print(f"FT6336U error: Invalid Chip ID (got {chip_id[0] if chip_id else None}, expected 0x64)")
self.initialized = False
return False
print(f"FT6336U touch controller detected (Chip ID: 0x{chip_id[0]:02X})")
# 2. Configure operating mode (0 = Normal Mode)
self.write_reg(self.REG_DEV_MODE, 0x00)
# 3. Configure CTRL mode (0 = Keep Active)
self.write_reg(self.REG_CTRL, self.CTRL_KEEP_ACTIVE)
self.initialized = True
return True
def is_touched(self):
"""Returns True if screen is touched by checking the TD_STATUS register and clearing coordinates."""
if not self.initialized:
return False
td_status = self.read_reg(self.REG_TD_STATUS)
if td_status is None:
return False
touch_count = td_status[0] & 0x0F
if touch_count > 0 and touch_count < 3:
# Read P1 coordinates to clear register/interrupt state on the chip
self.read_reg(self.REG_P1_XH, 6)
return True
return False
def read_touch(self):
"""Reads touch point coordinates.
Returns:
(x, y) coordinates of first touch point, or None if no touch.
"""
if not self.initialized:
return None
# Read TD_STATUS to check if there is an active touch
td_status = self.read_reg(self.REG_TD_STATUS)
if td_status is None:
return None
touch_count = td_status[0] & 0x0F
if touch_count == 0:
return None
# Read P1 coordinates (6 bytes starting at P1_XH)
buf = self.read_reg(self.REG_P1_XH, 6)
if buf is None or len(buf) < 6:
return None
raw_x = ((buf[0] & 0x0F) << 8) | buf[1]
raw_y = ((buf[2] & 0x0F) << 8) | buf[3]
# Apply coordinate transformations to map touch coordinate space to screen space
if self.swap_xy:
raw_x, raw_y = raw_y, raw_x
if self.invert_x:
raw_x = self.width - 1 - raw_x
if self.invert_y:
raw_y = self.height - 1 - raw_y
# Clamp coordinates to screen boundaries
x = max(0, min(self.width - 1, raw_x))
y = max(0, min(self.height - 1, raw_y))
return (x, y)
lib/ili9341.py
+261
View File
@@ -0,0 +1,261 @@
import time
from machine import Pin, SPI
import framebuf
import micropython
class ILI9341:
def __init__(self, spi, cs, dc, rst=None, bl=None, width=320, height=240, invert_color=True):
self.spi = spi
self.cs = cs
self.dc = dc
self.rst = rst
self.bl = bl
self.width = width
self.height = height
self.invert_color = invert_color
# 1. 1-bit Canvas Buffer (Standard MONO_HLSB for drawing)
self.hw_len = (self.width * self.height) // 8
self.canvas_buffer = bytearray(self.hw_len)
self.canvas = framebuf.FrameBuffer(self.canvas_buffer, self.width, self.height, framebuf.MONO_HLSB)
# Pre-allocate chunk buffer for conversion (16 rows: 320 * 16 * 2 = 10240 bytes)
self.chunk_rows = 16
self.row_buffer = bytearray(self.width * self.chunk_rows * 2)
# Initialize pins
self.cs.init(self.cs.OUT, value=1)
self.dc.init(self.dc.OUT, value=0)
if self.rst is not None:
self.rst.init(self.rst.OUT, value=1)
if self.bl is not None:
self.bl.init(self.bl.OUT, value=1)
self.reset()
self.init_display()
self.clear(0)
self.show()
# --- 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 ---
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())
src_fb = framebuf.FrameBuffer(data, w, h, framebuf.MONO_HLSB)
if scale == 1:
self.canvas.blit(src_fb, x, y)
else:
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):
if self.rst is not None:
self.rst(1); time.sleep_ms(5); self.rst(0); time.sleep_ms(15); self.rst(1); time.sleep_ms(15)
def write_cmd(self, cmd):
self.dc(0); self.cs(0); self.spi.write(bytearray([cmd])); self.cs(1)
def write_data(self, data):
self.dc(1); self.cs(0)
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):
# ILI9341 Initialization Sequence
self.write_cmd(0x01) # SWRESET
time.sleep_ms(150)
self.write_cmd(0xCF); self.write_data(b"\x00\xC1\x30")
self.write_cmd(0xED); self.write_data(b"\x64\x03\x12\x81")
self.write_cmd(0xE8); self.write_data(b"\x85\x00\x78")
self.write_cmd(0xCB); self.write_data(b"\x39\x2C\x00\x34\x02")
self.write_cmd(0xF7); self.write_data(b"\x20")
self.write_cmd(0xEA); self.write_data(b"\x00\x00")
self.write_cmd(0xC0); self.write_data(b"\x13") # Power Control 1
self.write_cmd(0xC1); self.write_data(b"\x13") # Power Control 2
self.write_cmd(0xC5); self.write_data(b"\x22\x35") # VCOM Control 1
self.write_cmd(0xC7); self.write_data(b"\xBD") # VCOM Control 2
# Memory Access Control (MADCTL) = 0x68 (Landscape: MV=1, MX=1, MY=0, BGR color filter)
self.write_cmd(0x36); self.write_data(b"\x68")
self.write_cmd(0xB6); self.write_data(b"\x0A\xA2") # Display Function Control
self.write_cmd(0x3A); self.write_data(b"\x55") # Pixel Format (COLMOD) = 16-bit RGB565
self.write_cmd(0xF6); self.write_data(b"\x01\x30")
self.write_cmd(0xB1); self.write_data(b"\x00\x1B") # Frame Rate Control
self.write_cmd(0xF2); self.write_data(b"\x00")
self.write_cmd(0x26); self.write_data(b"\x01") # Gamma Curve
self.write_cmd(0xE0); self.write_data(b"\x0F\x35\x31\x0B\x0E\x06\x49\xA7\x33\x07\x0F\x03\x0C\x0A\x00") # Positive Gamma Correction
self.write_cmd(0xE1); self.write_data(b"\x00\x0A\x0F\x04\x11\x08\x36\x58\x4D\x07\x10\x0C\x32\x34\x0F") # Negative Gamma Correction
if self.invert_color:
self.write_cmd(0x21) # INVON
else:
self.write_cmd(0x20) # INVOFF
self.write_cmd(0x11) # SLPOUT (Exit sleep mode)
time.sleep_ms(120)
self.write_cmd(0x29) # DISPON (Display on)
time.sleep_ms(10)
def invert(self, enable):
if enable:
self.write_cmd(0x21) # INVON
else:
self.write_cmd(0x20) # INVOFF
def set_window(self, x0, y0, x1, y1):
# Column Address Set (CASET)
self.write_cmd(0x2A)
self.write_data(bytearray([x0 >> 8, x0 & 0xFF, x1 >> 8, x1 & 0xFF]))
# Row Address Set (RASET)
self.write_cmd(0x2B)
self.write_data(bytearray([y0 >> 8, y0 & 0xFF, y1 >> 8, y1 & 0xFF]))
# Memory Write (RAMWR)
self.write_cmd(0x2C)
@micropython.native
def _convert_rows(self, start_row, num_rows, row_buf):
"""Converts 1-bit monochrome row segment to 16-bit RGB565 format.
Compiles block-wise bitwise operations at native speed.
"""
width = self.width
canvas_buf = self.canvas_buffer
idx = 0
for y in range(start_row, start_row + num_rows):
byte_offset = y * (width // 8)
for x_byte_idx in range(width // 8):
val = canvas_buf[byte_offset + x_byte_idx]
# Unroll 8 bits for speed
# Bit 7
if val & 0x80:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 6
if val & 0x40:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 5
if val & 0x20:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 4
if val & 0x10:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 3
if val & 0x08:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 2
if val & 0x04:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 1
if val & 0x02:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 0
if val & 0x01:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
def show(self):
"""Refreshes the screen by writing the frame buffer segment-by-segment."""
self.set_window(0, 0, self.width - 1, self.height - 1)
self.dc(1)
self.cs(0)
num_chunks = self.height // self.chunk_rows
for chunk in range(num_chunks):
start_row = chunk * self.chunk_rows
self._convert_rows(start_row, self.chunk_rows, self.row_buffer)
self.spi.write(self.row_buffer)
self.cs(1)
lib/rgb_led_util.py
+64
View File
@@ -0,0 +1,64 @@
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=None):
import sys
if pin_num is None:
pin_num = 14 if sys.platform == 'rp2' else 42
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()
lib/rlcd.py
+166
View File
@@ -0,0 +1,166 @@
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(0x21) # Inversion ON (Black Background by default)
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)
def invert(self, enable):
if enable:
self.write_cmd(0x21) # Inversion ON (Black Background)
else:
self.write_cmd(0x20) # Inversion OFF (White Background)
# --- 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)
lib/rtc_util.py
+122
View File
@@ -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"
lib/sd_util.py
+97
View File
@@ -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
lib/shtc3_util.py
+85
View File
@@ -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
lib/st7796.py
+249
View File
@@ -0,0 +1,249 @@
import time
from machine import Pin, SPI
import framebuf
import micropython
class ST7796:
def __init__(self, spi, cs, dc, rst, bl=None, width=480, height=320, invert_color=True):
self.spi = spi
self.cs = cs
self.dc = dc
self.rst = rst
self.bl = bl
self.width = width
self.height = height
self.invert_color = invert_color
# 1. 1-bit Canvas Buffer (Standard MONO_HLSB for drawing)
self.hw_len = (self.width * self.height) // 8
self.canvas_buffer = bytearray(self.hw_len)
self.canvas = framebuf.FrameBuffer(self.canvas_buffer, self.width, self.height, framebuf.MONO_HLSB)
# Pre-allocate chunk buffer for conversion (16 rows: 480 * 16 * 2 = 15360 bytes)
self.chunk_rows = 16
self.row_buffer = bytearray(self.width * self.chunk_rows * 2)
# Initialize 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)
if self.bl is not None:
# PWM or Pin backlight control
if isinstance(self.bl, Pin):
self.bl.init(self.bl.OUT, value=1)
self.reset()
self.init_display()
self.clear(0)
self.show()
# --- 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 ---
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())
src_fb = framebuf.FrameBuffer(data, w, h, framebuf.MONO_HLSB)
if scale == 1:
self.canvas.blit(src_fb, x, y)
else:
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(5); self.rst(0); time.sleep_ms(15); self.rst(1); time.sleep_ms(15)
def write_cmd(self, cmd):
self.dc(0); self.cs(0); self.spi.write(bytearray([cmd])); self.cs(1)
def write_data(self, data):
self.dc(1); self.cs(0)
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):
# ST7796 Minimal Initialization Sequence (ported from working C++ codebase)
self.write_cmd(0x01) # SWRESET
time.sleep_ms(150)
self.write_cmd(0x11) # SLPOUT
time.sleep_ms(120)
# Pixel Format (COLMOD) = 0x55 (16-bit color / RGB565)
self.write_cmd(0x3A); self.write_data(0x55)
time.sleep_ms(10)
# Memory Access Control (MADCTL) = 0xE0 (Landscape: MY=1, MX=1, MV=1, RGB order)
self.write_cmd(0x36); self.write_data(0xE0)
time.sleep_ms(10)
# Display Inversion Control
if self.invert_color:
self.write_cmd(0x21) # INVON
else:
self.write_cmd(0x20) # INVOFF
time.sleep_ms(10)
self.write_cmd(0x13) # NORON
time.sleep_ms(10)
self.write_cmd(0x29) # DISPON
time.sleep_ms(120)
def invert(self, enable):
if enable:
self.write_cmd(0x21) # INVON
else:
self.write_cmd(0x20) # INVOFF
def set_window(self, x0, y0, x1, y1):
# Column Address Set (CASET)
self.write_cmd(0x2A)
self.write_data(bytearray([x0 >> 8, x0 & 0xFF, x1 >> 8, x1 & 0xFF]))
# Row Address Set (RASET)
self.write_cmd(0x2B)
self.write_data(bytearray([y0 >> 8, y0 & 0xFF, y1 >> 8, y1 & 0xFF]))
# Memory Write (RAMWR)
self.write_cmd(0x2C)
@micropython.native
def _convert_rows(self, start_row, num_rows, row_buf):
"""Converts 1-bit monochrome row segment to 16-bit RGB565 format.
Compiles block-wise bitwise operations at native speed.
"""
width = self.width
canvas_buf = self.canvas_buffer
idx = 0
for y in range(start_row, start_row + num_rows):
byte_offset = y * (width // 8)
for x_byte_idx in range(width // 8):
val = canvas_buf[byte_offset + x_byte_idx]
# Unroll 8 bits for speed
# Bit 7
if val & 0x80:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 6
if val & 0x40:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 5
if val & 0x20:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 4
if val & 0x10:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 3
if val & 0x08:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 2
if val & 0x04:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 1
if val & 0x02:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
# Bit 0
if val & 0x01:
row_buf[idx] = 0xFF; row_buf[idx+1] = 0xFF
else:
row_buf[idx] = 0x00; row_buf[idx+1] = 0x00
idx += 2
def show(self):
"""Refreshes the screen by writing the frame buffer segment-by-segment."""
self.set_window(0, 0, self.width - 1, self.height - 1)
self.dc(1)
self.cs(0)
num_chunks = self.height // self.chunk_rows
for chunk in range(num_chunks):
start_row = chunk * self.chunk_rows
self._convert_rows(start_row, self.chunk_rows, self.row_buffer)
self.spi.write(self.row_buffer)
self.cs(1)