a2nr
/
velxio
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor I2C slave tests for ESP32: update event handling and improve accuracy of ACK/NACK responses; add full end-to-end test for MPU-6050 I2C simulation; update components metadata timestamp; mark subproject commits as dirty for wokwi-libs.

This commit is contained in:
David Montero Crespo 2026-04-09 15:06:39 -03:00
parent 5795b1d506
commit 46e459f51b
5 changed files with 649 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
backend/app/services/esp32_i2c_slaves.py
View File

@ -2,33 +2,34 @@
esp32_i2c_slaves.py — Standalone I2C slave state machines for ESP32 QEMU simulation.
Each class emulates the I2C register map of a real sensor, handling the picsimlab
I2C event protocol:
I2C event protocol as defined in hw/i2c/picsimlab_i2c.c:
event & 0x00FF (LOW byte) = operation type:
0x01 = START — return 1 (ACK = device present)
0x05 = WRITE — (byte in bits 15:8) return 1 (ACK)
0x06 = WRITE — (byte in bits 15:8) return 1 (ACK) — continuation byte
0x03 = READ — return register byte at current pointer
0x00 = STOP
picsimlab_i2c_ev(event) → passes raw QEMU i2c_event enum value:
0x00 = I2C_START_RECV — firmware doing requestFrom (read direction START)
0x01 = I2C_START_SEND — firmware doing beginTransmission (write direction START)
0x02 = I2C_START_SEND_ASYNC (rarely used)
0x03 = I2C_FINISH — end of transaction (STOP or RSTART between write+read)
0x04 = I2C_NACK
Data byte for WRITE: (event >> 8) & 0xFF (HIGH byte)
picsimlab_i2c_tx(data) → event = (data << 8) | (I2C_NACK+1) = (data<<8)|0x05
picsimlab_i2c_rx() → event = I2C_NACK+2 = 0x06 (return data byte to firmware)
These classes are imported by esp32_worker.py and by test_esp32_i2c_slaves.py.
ACK convention (matches QEMU i2c core):
return 0 → ACK (success, device present / byte accepted)
return ≠0 → NACK (error)
For READ events: return value is the data byte delivered to the firmware.
"""
import datetime as _datetime
# ── Protocol constants (actual picsimlab encoding) ────────────────────────────
# ── Protocol constants ────────────────────────────────────────────────────────
I2C_STOP = 0x00 # event & 0xFF
I2C_START = 0x01 # event & 0xFF
I2C_READ = 0x03 # event & 0xFF
# WRITE uses two codes:
# 0x05 = first byte in a write burst (register address)
# 0x06 = subsequent byte in a write burst (data)
# Both are handled identically by slaves — first_byte flag distinguishes address vs data.
_I2C_WRITE_CODES = (0x05, 0x06)
I2C_START_RECV = 0x00 # firmware called requestFrom (read direction START)
I2C_START_SEND = 0x01 # firmware called beginTransmission (write direction START)
I2C_FINISH = 0x03 # end of transaction (STOP or repeated-START between phases)
I2C_WRITE = 0x05 # firmware sent a byte; data = (event >> 8) & 0xFF
I2C_READ = 0x06 # firmware requesting a byte; return the data byte
# ── MPU-6050 IMU ──────────────────────────────────────────────────────────────
@ -37,19 +38,10 @@ class MPU6050Slave:
"""Full MPU-6050 register-map I2C slave emulation (address 0x68 or 0x69)."""
def __init__(self, addr: int = 0x68):
self.addr = addr
self.regs = bytearray(256)
# Default reg_ptr to WHO_AM_I so the first READ (without a preceding
# WRITE, as happens with Adafruit BusIO write-then-read) returns 0x68.
self.reg_ptr = 0x75
self.addr = addr
self.regs = bytearray(256)
self.reg_ptr = 0
self.first_byte = True
# Adafruit_I2CDevice::begin() fires TWO WHO_AM_I reads before the
# library moves on to actual data reads:
# 1. detected() uses requestFrom as fallback — fires START+READ
# 2. chip_id_register.read() — fires START+READ
# Only after both have returned 0x68 do we switch reg_ptr to 0x3B
# (start of accel/gyro/temp block) for subsequent data transactions.
self._who_am_i_count = 0
# WHO_AM_I
self.regs[0x75] = 0x68
@ -72,42 +64,35 @@ class MPU6050Slave:
op = event & 0xFF # low byte = operation type
data = (event >> 8) & 0xFF # high byte = data byte (for WRITE)
if op == I2C_START:
if op in (I2C_START_RECV, I2C_START_SEND):
# New transaction beginning. Reset first_byte flag.
# reg_ptr is NOT reset here — a write-then-read (repeated START)
# relies on reg_ptr having been set by the preceding WRITE phase.
self.first_byte = True
# picsimlab does not fire WRITE callbacks for write-then-read
# transactions (endTransmission(false) + requestFrom).
# Adafruit_MPU6050::begin() fires THREE START+READ sequences
# before any data reads:
# 1. Wire.begin(sda, scl) bus-init probe → START+READ(WHO_AM_I)
# 2. _wire->begin() inside i2c_dev->begin() → START+READ(WHO_AM_I)
# 3. chip_id_register.read() → START+READ(WHO_AM_I) ← must still return 0x68
# Only after all three have returned 0x68 do we switch to data mode
# (so subsequent reset() + getEvent() READs get accel/gyro bytes).
if self._who_am_i_count >= 3:
self.reg_ptr = 0x3B # sensor data block
else:
self.reg_ptr = 0x75 # WHO_AM_I register
return 1 # ACK — device present
elif op in _I2C_WRITE_CODES:
return 0 # ACK (0 = success in QEMU convention)
elif op == I2C_WRITE:
if self.first_byte:
# First byte after START is the register address pointer
self.reg_ptr = data
self.first_byte = False
else:
# Subsequent bytes are data written into the register map
self.regs[self.reg_ptr] = data
# Auto-clear DEVICE_RESET bit (reg 0x6B bit 7) so the
# Adafruit begin() reset-wait loop exits immediately.
# Auto-clear DEVICE_RESET bit (bit 7 of PWR_MGMT_1 = 0x6B)
# so the Adafruit begin() reset-wait loop exits immediately.
if self.reg_ptr == 0x6B:
self.regs[0x6B] &= 0x7F
self.reg_ptr = (self.reg_ptr + 1) & 0xFF
return 1 # ACK
return 0 # ACK
elif op == I2C_READ:
# Return the byte at the current register pointer, then advance it.
val = self.regs[self.reg_ptr]
# Track WHO_AM_I reads to know when begin() has confirmed device
if self.reg_ptr == 0x75 and val == 0x68:
self._who_am_i_count += 1
self.reg_ptr = (self.reg_ptr + 1) & 0xFF
return val
else: # STOP / unknown
else: # I2C_FINISH, I2C_NACK, unknown
self.first_byte = True
return 0
@ -234,15 +219,15 @@ class BMP280Slave:
op = event & 0xFF
data = (event >> 8) & 0xFF
if op == I2C_START:
self.first_byte = True; return 1
elif op in _I2C_WRITE_CODES:
if op in (I2C_START_RECV, I2C_START_SEND):
self.first_byte = True; return 0
elif op == I2C_WRITE:
if self.first_byte:
self.reg_ptr = data; self.first_byte = False
else:
self.regs[self.reg_ptr] = data
self.reg_ptr = (self.reg_ptr + 1) & 0xFF
return 1
return 0
elif op == I2C_READ:
val = self.regs[self.reg_ptr]
self.reg_ptr = (self.reg_ptr + 1) & 0xFF
@ -279,12 +264,12 @@ class DS1307Slave:
op = event & 0xFF
data = (event >> 8) & 0xFF
if op == I2C_START:
self.first_byte = True; return 1
elif op in _I2C_WRITE_CODES:
if op in (I2C_START_RECV, I2C_START_SEND):
self.first_byte = True; return 0
elif op == I2C_WRITE:
if self.first_byte:
self.reg_ptr = data; self.first_byte = False
return 1
return 0
elif op == I2C_READ:
val = self._read_reg(self.reg_ptr)
self.reg_ptr = (self.reg_ptr + 1) & 0x3F
@ -314,7 +299,7 @@ class DS3231Slave(DS1307Slave):
# ── I2C Write Sink (relay for write-only devices: SSD1306, PCF8574) ──────────
class I2CWriteSink:
"""ACKs all I2C writes, emits complete transaction to frontend on STOP."""
"""ACKs all I2C writes, emits complete transaction to frontend on FINISH."""
def __init__(self, addr: int, emit_fn) -> None:
self.addr = addr
@ -325,13 +310,13 @@ class I2CWriteSink:
op = event & 0xFF
data = (event >> 8) & 0xFF
if op == I2C_START: # START — reset buffer
self._buf = []; return 1
elif op in _I2C_WRITE_CODES: # WRITE — accumulate byte
self._buf.append(data); return 1
elif op == I2C_READ: # READ — write-only device
return 0xFF
else: # STOP — emit transaction
if op in (I2C_START_RECV, I2C_START_SEND):
self._buf = []; return 0
elif op == I2C_WRITE:
self._buf.append(data); return 0
elif op == I2C_READ:
return 0xFF # write-only device
else: # I2C_FINISH — emit accumulated transaction
if self._buf:
self._emit({'type': 'i2c_transaction',
'addr': self.addr, 'data': list(self._buf)})

55
backend/app/services/esp32_worker.py
View File

@ -581,17 +581,59 @@ def main() -> None: # noqa: C901 (complexity OK for inline worker)
if pixels:
_emit({'type': 'ws2812_update', 'channel': channel, 'pixels': pixels})
# ── Per-slave I2C event counter (for logging) ─────────────────────────────
_i2c_event_seq: dict = {} # addr → event count
_I2C_OP_NAME = {0x00: 'START_RECV', 0x01: 'START_SEND', 0x02: 'START_ASYNC',
0x03: 'FINISH', 0x04: 'NACK',
0x05: 'WRITE', 0x06: 'READ'}
_MPU_REG_NAME = {
0x19: 'SMPRT_DIV', 0x1A: 'CONFIG', 0x1B: 'GYRO_CFG', 0x1C: 'ACCEL_CFG',
0x3B: 'AX_H', 0x3C: 'AX_L', 0x3D: 'AY_H', 0x3E: 'AY_L',
0x3F: 'AZ_H', 0x40: 'AZ_L', 0x41: 'T_H', 0x42: 'T_L',
0x43: 'GX_H', 0x44: 'GX_L', 0x45: 'GY_H', 0x46: 'GY_L',
0x47: 'GZ_H', 0x48: 'GZ_L',
0x6B: 'PWR_MGMT1', 0x68: 'SIG_RST', 0x75: 'WHO_AM_I',
}
def _on_i2c_event(bus_id: int, addr: int, event: int) -> int:
"""Synchronous — must return immediately; called from QEMU thread."""
# Register-map slaves (MPU-6050, etc.) take priority over static responses
slave = _i2c_slaves.get(addr)
op = event & 0xFF
data = (event >> 8) & 0xFF
op = event & 0xFF
data = (event >> 8) & 0xFF
op_name = _I2C_OP_NAME.get(op, f'0x{op:02x}')
if slave is not None:
result = slave.handle_event(event)
_log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} op=0x{op:02x} data=0x{data:02x} result=0x{result:02x} slave={type(slave).__name__} reg_ptr=0x{getattr(slave,"reg_ptr",0):02x}')
result = slave.handle_event(event)
reg_ptr = getattr(slave, 'reg_ptr', 0)
# Build descriptive annotation
if op in (0x00, 0x01): # START_RECV / START_SEND
note = f'→ reg_ptr=0x{reg_ptr:02x}'
elif op == 0x06: # READ byte (actual data delivery to firmware)
reg_nm = _MPU_REG_NAME.get((reg_ptr - 1) & 0xFF, f'0x{(reg_ptr-1)&0xFF:02x}')
note = f'→ {reg_nm}=0x{result:02x}'
elif op == 0x05: # WRITE byte
note = f'byte=0x{data:02x} → reg_ptr=0x{reg_ptr:02x}'
else:
note = ''
if type(slave).__name__ == 'MPU6050Slave':
seq = _i2c_event_seq
n = seq[addr] = seq.get(addr, 0) + 1
_log(f'I2C #{n:03d} bus={bus_id} addr=0x{addr:02x} {op_name} {note}')
else:
_log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} '
f'op={op_name} result=0x{result:02x} slave={type(slave).__name__}')
# Emit trace event to WebSocket so JS test can observe I2C traffic
if not _stopped.is_set():
_emit({'type': 'i2c_trace', 'bus': bus_id, 'addr': addr,
'event': event, 'op': op_name, 'result': result,
'reg_ptr': reg_ptr})
return result
_log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} op=0x{op:02x} NO_SLAVE registered={list(_i2c_slaves.keys())}')
_log(f'I2C bus={bus_id} addr=0x{addr:02x} event=0x{event:04x} op={op_name} '
f'NO_SLAVE registered={list(_i2c_slaves.keys())}')
resp = _i2c_responses.get(addr, 0)
if not _stopped.is_set():
_emit({'type': 'i2c_event', 'bus': bus_id, 'addr': addr,
@ -711,7 +753,6 @@ def main() -> None: # noqa: C901 (complexity OK for inline worker)
sensor_data['slave'] = sink
_sensors[gpio] = sensor_data
_sensors_ready.set()
_log(f'esp32_i2c_slaves: MPU6050Slave default reg_ptr=0x{_MPU6050Slave().reg_ptr:02x} (expect 0x75)')
_log(f'_i2c_slaves registered: {list(_i2c_slaves.keys())}')
_emit({'type': 'system', 'event': 'booted'})

258
backend/test_esp32_i2c_slaves.py
View File

@ -1,18 +1,19 @@
"""
Tests for ESP32 I2C slave state machines.
Covers BMP280Slave, DS1307Slave, DS3231Slave, and I2CWriteSink from
app/services/esp32_i2c_slaves.py — the Python register-map emulators
that handle I2C traffic from QEMU-simulated ESP32 firmware.
Covers BMP280Slave, DS1307Slave, DS3231Slave, I2CWriteSink, and MPU6050Slave
from app/services/esp32_i2c_slaves.py.
Actual picsimlab I2C event encoding (confirmed by observing real QEMU events):
event & 0xFF = operation type:
0x01 = START → slave must return 1 (ACK = device present)
0x05 = WRITE (first byte / register address) → slave returns 1 (ACK)
0x06 = WRITE (subsequent bytes / data) → slave returns 1 (ACK)
0x03 = READ → slave returns register byte at current pointer
0x00 = STOP → slave resets state
(event >> 8) & 0xFF = data byte (for WRITE events)
Correct picsimlab I2C event encoding (from hw/i2c/picsimlab_i2c.c + QEMU i2c.h):
event & 0xFF = operation type:
0x00 = I2C_START_RECV — firmware called requestFrom (read direction START)
0x01 = I2C_START_SEND — firmware called beginTransmission (write direction START)
0x03 = I2C_FINISH — end of transaction (STOP or RSTART between write+read)
0x05 = WRITE byte — (event >> 8) & 0xFF is the data byte
0x06 = READ byte — return value is the data byte to deliver to firmware
ACK convention: return 0 = ACK (success), non-zero = NACK.
For READ events: return value is the data byte sent to the firmware.
Run from the backend/ directory:
python test_esp32_i2c_slaves.py
@ -31,27 +32,33 @@ from app.services.esp32_i2c_slaves import (
DS3231Slave,
I2CWriteSink,
MPU6050Slave,
I2C_START_RECV,
I2C_START_SEND,
I2C_FINISH,
I2C_WRITE,
I2C_READ,
)
# ── I2C protocol helpers (correct picsimlab encoding) ─────────────────────────
I2C_START = 0x0001
I2C_STOP = 0x0000
I2C_READ = 0x0003
# ── I2C protocol helpers ──────────────────────────────────────────────────────
def i2c_write(byte: int) -> int:
"""WRITE event: data in high byte, type 0x05 in low byte."""
return ((byte & 0xFF) << 8) | 0x05
"""WRITE event: data in high byte, op 0x05 in low byte."""
return ((byte & 0xFF) << 8) | I2C_WRITE
def i2c_read_seq(slave, reg: int, n: int) -> list[int]:
"""Set register pointer then read n bytes sequentially."""
slave.handle_event(I2C_START)
slave.handle_event(i2c_write(reg))
"""Simulate write-then-read: write register address, then read n bytes.
Models the Adafruit BusIO write_then_read pattern:
beginTransmission → write(reg) → endTransmission(false) → requestFrom → read()*n
"""
slave.handle_event(I2C_START_SEND) # write direction START
slave.handle_event(i2c_write(reg)) # set register pointer
slave.handle_event(I2C_FINISH) # RSTART (repeated start before read phase)
slave.handle_event(I2C_START_RECV) # read direction START
data = [slave.handle_event(I2C_READ) for _ in range(n)]
slave.handle_event(I2C_STOP)
slave.handle_event(I2C_FINISH) # STOP
return data
@ -76,18 +83,22 @@ class TestBMP280Slave(unittest.TestCase):
# ── I2C protocol ───────────────────────────────────────────────────────────
def test_ack_on_start(self):
result = self.slave.handle_event(I2C_START)
self.assertEqual(result, 1, 'START must return 1 (device present)')
def test_ack_on_start_send(self):
result = self.slave.handle_event(I2C_START_SEND)
self.assertEqual(result, 0, 'START_SEND must return 0 (ACK)')
def test_ack_on_start_recv(self):
result = self.slave.handle_event(I2C_START_RECV)
self.assertEqual(result, 0, 'START_RECV must return 0 (ACK)')
def test_ack_on_write(self):
self.slave.handle_event(I2C_START)
self.slave.handle_event(I2C_START_SEND)
result = self.slave.handle_event(i2c_write(0xD0))
self.assertEqual(result, 1, 'WRITE must return 1 (ACK)')
self.assertEqual(result, 0, 'WRITE must return 0 (ACK)')
def test_stop_returns_zero(self):
self.slave.handle_event(I2C_START)
result = self.slave.handle_event(I2C_STOP)
def test_finish_returns_zero(self):
self.slave.handle_event(I2C_START_SEND)
result = self.slave.handle_event(I2C_FINISH)
self.assertEqual(result, 0)
# ── Chip identity ──────────────────────────────────────────────────────────
@ -161,21 +172,26 @@ class TestBMP280Slave(unittest.TestCase):
# ── State machine ─────────────────────────────────────────────────────────
def test_write_sets_register_ptr(self):
self.slave.handle_event(I2C_START)
"""Write-then-read: write reg address 0xD0, then read returns chip_id."""
self.slave.handle_event(I2C_START_SEND)
self.slave.handle_event(i2c_write(0xD0)) # set ptr to chip_id reg
self.slave.handle_event(I2C_FINISH)
self.slave.handle_event(I2C_START_RECV)
val = self.slave.handle_event(I2C_READ)
self.assertEqual(val, 0x60)
def test_stop_resets_first_byte_flag(self):
"""After STOP, the next transaction's first WRITE must set reg_ptr, not write data."""
self.slave.handle_event(I2C_START)
def test_finish_resets_first_byte_flag(self):
"""After FINISH, the next transaction's first WRITE must set reg_ptr."""
self.slave.handle_event(I2C_START_SEND)
self.slave.handle_event(i2c_write(0xD0))
self.slave.handle_event(I2C_STOP)
# New transaction: WRITE 0xD0 again → should set reg_ptr, not write to 0xD0
self.slave.handle_event(I2C_START)
self.slave.handle_event(I2C_FINISH)
# New transaction: WRITE 0xD0 again → should set reg_ptr, not write data
self.slave.handle_event(I2C_START_SEND)
self.slave.handle_event(i2c_write(0xD0))
self.slave.handle_event(I2C_FINISH)
self.slave.handle_event(I2C_START_RECV)
val = self.slave.handle_event(I2C_READ)
self.assertEqual(val, 0x60, 'chip_id should still be 0x60 after STOP + new transaction')
self.assertEqual(val, 0x60, 'chip_id should still be 0x60 after FINISH + new transaction')
# ══════════════════════════════════════════════════════════════════════════════
@ -187,12 +203,12 @@ class TestDS1307Slave(unittest.TestCase):
def setUp(self):
self.slave = DS1307Slave()
def test_ack_on_start(self):
self.assertEqual(self.slave.handle_event(I2C_START), 1)
def test_ack_on_start_send(self):
self.assertEqual(self.slave.handle_event(I2C_START_SEND), 0)
def test_ack_on_write(self):
self.slave.handle_event(I2C_START)
self.assertEqual(self.slave.handle_event(i2c_write(0x00)), 1)
self.slave.handle_event(I2C_START_SEND)
self.assertEqual(self.slave.handle_event(i2c_write(0x00)), 0)
def test_seconds_is_valid_bcd(self):
seconds = i2c_read_seq(self.slave, 0x00, 1)[0]
@ -293,65 +309,68 @@ class TestI2CWriteSink(unittest.TestCase):
self.emitted: list[dict] = []
self.sink = I2CWriteSink(addr=0x3C, emit_fn=self.emitted.append)
def test_ack_on_start(self):
self.assertEqual(self.sink.handle_event(I2C_START), 1)
def test_ack_on_start_send(self):
self.assertEqual(self.sink.handle_event(I2C_START_SEND), 0)
def test_ack_on_start_recv(self):
self.assertEqual(self.sink.handle_event(I2C_START_RECV), 0)
def test_ack_on_write(self):
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
for byte in [0x00, 0x21, 0xAB]:
result = self.sink.handle_event(i2c_write(byte))
self.assertEqual(result, 1, f'WRITE 0x{byte:02X} must return 1 (ACK)')
self.assertEqual(result, 0, f'WRITE 0x{byte:02X} must return 0 (ACK)')
def test_read_returns_0xff(self):
"""Write-only device: READ must return 0xFF (no data to send)."""
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
result = self.sink.handle_event(I2C_READ)
self.assertEqual(result, 0xFF)
def test_emits_on_stop(self):
"""After 3 writes + STOP, emit_fn must be called exactly once."""
self.sink.handle_event(I2C_START)
def test_emits_on_finish(self):
"""After 3 writes + FINISH, emit_fn must be called exactly once."""
self.sink.handle_event(I2C_START_SEND)
for b in [0x00, 0x21, 0x7F]:
self.sink.handle_event(i2c_write(b))
self.sink.handle_event(I2C_STOP)
self.assertEqual(len(self.emitted), 1, 'emit_fn must be called once on STOP')
self.sink.handle_event(I2C_FINISH)
self.assertEqual(len(self.emitted), 1, 'emit_fn must be called once on FINISH')
def test_emit_payload_addr(self):
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
self.sink.handle_event(i2c_write(0xAB))
self.sink.handle_event(I2C_STOP)
self.sink.handle_event(I2C_FINISH)
self.assertEqual(self.emitted[0]['addr'], 0x3C)
def test_emit_payload_type(self):
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
self.sink.handle_event(i2c_write(0xAB))
self.sink.handle_event(I2C_STOP)
self.sink.handle_event(I2C_FINISH)
self.assertEqual(self.emitted[0]['type'], 'i2c_transaction')
def test_emit_payload_data(self):
bytes_ = [0x00, 0x21, 0x7F]
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
for b in bytes_:
self.sink.handle_event(i2c_write(b))
self.sink.handle_event(I2C_STOP)
self.sink.handle_event(I2C_FINISH)
self.assertEqual(self.emitted[0]['data'], bytes_)
def test_no_emit_for_empty_buffer(self):
"""START then immediate STOP with no writes must NOT emit."""
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_STOP)
"""START then immediate FINISH with no writes must NOT emit."""
self.sink.handle_event(I2C_START_SEND)
self.sink.handle_event(I2C_FINISH)
self.assertEqual(len(self.emitted), 0, 'Empty buffer must not trigger emit_fn')
def test_resets_buffer_after_emit(self):
"""Second transaction accumulates fresh bytes, not leftover from first."""
# First transaction: writes [0xAA]
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
self.sink.handle_event(i2c_write(0xAA))
self.sink.handle_event(I2C_STOP)
self.sink.handle_event(I2C_FINISH)
# Second transaction: writes [0xBB]
self.sink.handle_event(I2C_START)
self.sink.handle_event(I2C_START_SEND)
self.sink.handle_event(i2c_write(0xBB))
self.sink.handle_event(I2C_STOP)
self.sink.handle_event(I2C_FINISH)
self.assertEqual(len(self.emitted), 2)
self.assertEqual(self.emitted[0]['data'], [0xAA])
self.assertEqual(self.emitted[1]['data'], [0xBB])
@ -359,15 +378,15 @@ class TestI2CWriteSink(unittest.TestCase):
def test_custom_addr_forwarded(self):
"""Sink created with addr=0x27 emits with that addr."""
sink = I2CWriteSink(addr=0x27, emit_fn=self.emitted.append)
sink.handle_event(I2C_START)
sink.handle_event(I2C_START_SEND)
sink.handle_event(i2c_write(0x38))
sink.handle_event(I2C_STOP)
sink.handle_event(I2C_FINISH)
self.assertEqual(self.emitted[0]['addr'], 0x27)
def test_stop_return_value(self):
"""STOP always returns 0."""
self.sink.handle_event(I2C_START)
result = self.sink.handle_event(I2C_STOP)
def test_finish_return_value(self):
"""FINISH always returns 0."""
self.sink.handle_event(I2C_START_SEND)
result = self.sink.handle_event(I2C_FINISH)
self.assertEqual(result, 0)
@ -385,47 +404,32 @@ class TestMPU6050Slave(unittest.TestCase):
result = i2c_read_seq(self.mpu, 0x75, 1)
self.assertEqual(result[0], 0x68)
def test_start_ack(self):
"""START event must return 1 (ACK — device present)."""
self.assertEqual(self.mpu.handle_event(I2C_START), 1)
def test_start_send_ack(self):
"""START_SEND event must return 0 (ACK)."""
self.assertEqual(self.mpu.handle_event(I2C_START_SEND), 0)
def test_begin_no_write_events(self):
"""Simulates Adafruit BusIO write-then-read: START then READ, no WRITE.
picsimlab does not fire WRITE callbacks for write-then-read transactions,
so the slave must return WHO_AM_I (0x68) on the first READ regardless."""
def test_start_recv_ack(self):
"""START_RECV event must return 0 (ACK)."""
self.assertEqual(self.mpu.handle_event(I2C_START_RECV), 0)
def test_write_then_read_who_am_i(self):
"""Full write-then-read: write reg 0x75, then read returns 0x68."""
m = MPU6050Slave()
self.assertEqual(m.handle_event(I2C_START), 1) # START → ACK
result = m.handle_event(I2C_READ) # READ without prior WRITE
self.assertEqual(result, 0x68,
f"Expected WHO_AM_I=0x68 without WRITE, got 0x{result:02x}")
m.handle_event(I2C_START_SEND) # beginTransmission
m.handle_event(i2c_write(0x75)) # write register address
m.handle_event(I2C_FINISH) # RSTART
m.handle_event(I2C_START_RECV) # requestFrom
result = m.handle_event(I2C_READ) # read byte
self.assertEqual(result, 0x68, f"WHO_AM_I must be 0x68, got 0x{result:02x}")
def test_data_read_after_begin(self):
"""After begin() succeeds (three WHO_AM_I reads), START resets reg_ptr to 0x3B.
Adafruit_MPU6050::begin() fires THREE START+READ sequences before data reads:
1. Wire.begin(sda, scl) bus-init probe → START+READ(WHO_AM_I)
2. _wire->begin() inside i2c_dev->begin() → START+READ(WHO_AM_I)
3. chip_id_register.read() → START+READ(WHO_AM_I) ← must still return 0x68
Only after all three return 0x68 does the slave switch to data mode.
"""
def test_detected_pattern(self):
"""detected() pattern: START_SEND then FINISH (no data bytes) returns ACK."""
m = MPU6050Slave()
# First WHO_AM_I read: Wire.begin(sda, scl) bus-init probe
m.handle_event(I2C_START)
r1 = m.handle_event(I2C_READ)
self.assertEqual(r1, 0x68, "First WHO_AM_I read must return 0x68")
# Second WHO_AM_I read: _wire->begin() inside i2c_dev->begin()
m.handle_event(I2C_START)
r2 = m.handle_event(I2C_READ)
self.assertEqual(r2, 0x68, "Second WHO_AM_I read must return 0x68")
# Third WHO_AM_I read: chip_id_register.read() — count reaches 3
m.handle_event(I2C_START)
r3 = m.handle_event(I2C_READ)
self.assertEqual(r3, 0x68, "Third WHO_AM_I read (chip_id check) must return 0x68")
# Now _who_am_i_count=3 → next START switches to data mode
m.handle_event(I2C_START)
first_accel_byte = m.handle_event(I2C_READ)
self.assertEqual(first_accel_byte, m.regs[0x3B],
"After three WHO_AM_I reads, START should reset reg_ptr to 0x3B (accel block)")
# beginTransmission + endTransmission (no data)
r1 = m.handle_event(I2C_START_SEND)
self.assertEqual(r1, 0, "START_SEND must return 0 (ACK) for detected()")
r2 = m.handle_event(I2C_FINISH)
self.assertEqual(r2, 0, "FINISH must return 0")
def test_accel_z_default_1g(self):
"""ACCEL_Z should default to +1g = 0x4000 (MSB=0x40, LSB=0x00)."""
@ -441,12 +445,20 @@ class TestMPU6050Slave(unittest.TestCase):
# 1g at ±2g full-scale = 16384 = 0x4000
self.assertEqual(accel_x, 0x4000)
def test_sequential_read_14_bytes(self):
"""getEvent() reads 14 bytes from 0x3B — all must come from correct regs."""
result = i2c_read_seq(self.mpu, 0x3B, 14)
self.assertEqual(len(result), 14)
expected = list(self.mpu.regs[0x3B:0x3B + 14])
self.assertEqual(result, expected, "14-byte read from 0x3B must match register map")
def test_device_reset_bit_auto_cleared(self):
"""Writing 0x80 to PWR_MGMT_1 (0x6B) must auto-clear bit 7 immediately."""
self.mpu.handle_event(I2C_START)
self.mpu.handle_event(i2c_write(0x6B)) # register address
self.mpu.handle_event(0x8006) # write 0x80 (DEVICE_RESET)
# Read back: bit 7 should be 0 (reset complete)
self.mpu.handle_event(I2C_START_SEND)
self.mpu.handle_event(i2c_write(0x6B)) # register address
self.mpu.handle_event(i2c_write(0x80)) # write 0x80 (DEVICE_RESET bit)
self.mpu.handle_event(I2C_FINISH)
# Read back: bit 7 should be 0 (auto-cleared)
result = i2c_read_seq(self.mpu, 0x6B, 1)
self.assertEqual(result[0] & 0x80, 0, "DEVICE_RESET bit must auto-clear")
@ -456,6 +468,26 @@ class TestMPU6050Slave(unittest.TestCase):
result = i2c_read_seq(mpu69, 0x75, 1)
self.assertEqual(result[0], 0x68)
def test_reg_ptr_preserved_across_rstart(self):
"""Write sets reg_ptr; FINISH then START_RECV should NOT reset reg_ptr."""
m = MPU6050Slave()
m.handle_event(I2C_START_SEND)
m.handle_event(i2c_write(0x75)) # set reg_ptr = 0x75
m.handle_event(I2C_FINISH) # RSTART — must NOT reset reg_ptr
m.handle_event(I2C_START_RECV) # read direction START
val = m.handle_event(I2C_READ)
self.assertEqual(val, 0x68, "reg_ptr must be preserved across RSTART")
def test_write_to_reg(self):
"""Writing two bytes sets reg address then reg value."""
m = MPU6050Slave()
m.handle_event(I2C_START_SEND)
m.handle_event(i2c_write(0x6B)) # reg address
m.handle_event(i2c_write(0x01)) # value
m.handle_event(I2C_FINISH)
result = i2c_read_seq(m, 0x6B, 1)
self.assertEqual(result[0], 0x01)
# ── Runner ────────────────────────────────────────────────────────────────────

401
backend/test_mpu6050_simulation.mjs Normal file
View File

@ -0,0 +1,401 @@
/**
* test_mpu6050_simulation.mjs
*
* Full end-to-end test for the ESP32 + MPU-6050 I2C simulation.
* Mirrors exactly what the frontend does:
* 1. POST /api/compile/ → get firmware_b64
* 2. WebSocket /api/simulation/ws/{id}
* 3. send start_esp32 with firmware + sensors:[{sensor_type:'mpu6050',…}]
* 4. Watch all events, log I2C state machine + serial output
*
* Run from the backend/ directory:
* node test_mpu6050_simulation.mjs [--timeout 30]
*
* Prerequisites: Backend running on http://localhost:8001
*/
// ─── Config ───────────────────────────────────────────────────────────────────
const BACKEND = process.env.BACKEND_URL ?? process.argv.find(a => a.startsWith('--backend='))?.slice(10) ?? 'http://localhost:8001';
const WS_BASE = BACKEND.replace(/^https?:/, m => m === 'https:' ? 'wss:' : 'ws:');
const SESSION = `test-mpu6050-${Date.now()}`;
const TIMEOUT_S = parseInt(process.argv.find(a => a.startsWith('--timeout='))?.slice(10) ?? '40');
// ─── MPU-6050 sketch (same as the example in examples.ts) ────────────────────
const SKETCH = `// ESP32 — MPU-6050 Accelerometer & Gyroscope (I2C)
// Requires: Adafruit MPU6050, Adafruit Unified Sensor libraries
// Wiring: SDA → D21 | SCL → D22 | VCC → 3V3 | GND → GND
#include <Adafruit_MPU6050.h>
#include <Adafruit_Sensor.h>
#include <Wire.h>
Adafruit_MPU6050 mpu;
void setup() {
Serial.begin(115200);
Wire.begin(21, 22); // SDA=21, SCL=22
if (!mpu.begin()) {
Serial.println("MPU6050 not found! Check wiring.");
while (true) delay(10);
}
mpu.setAccelerometerRange(MPU6050_RANGE_8_G);
mpu.setGyroRange(MPU6050_RANGE_500_DEG);
mpu.setFilterBandwidth(MPU6050_BAND_21_HZ);
Serial.println("MPU6050 ready!");
}
void loop() {
sensors_event_t a, g, temp;
mpu.getEvent(&a, &g, &temp);
Serial.printf("Accel X=%.2f Y=%.2f Z=%.2f m/s^2\\n",
a.acceleration.x, a.acceleration.y, a.acceleration.z);
Serial.printf("Gyro X=%.2f Y=%.2f Z=%.2f rad/s\\n",
g.gyro.x, g.gyro.y, g.gyro.z);
Serial.printf("Temp: %.1f C\\n---\\n", temp.temperature);
delay(500);
}`;
// ─── I2C event decoder (mirrors Python esp32_i2c_slaves.py constants) ─────────
const I2C_OP = { 0x00: 'STOP', 0x01: 'START', 0x03: 'READ', 0x05: 'WRITE_FIRST', 0x06: 'WRITE_CONT' };
// Local mirror of the MPU6050 slave state machine — tracks the same logic as
// Python's MPU6050Slave so we can annotate WHICH firmware call each event came from.
class MPU6050StateMirror {
constructor() {
this.reg_ptr = 0x75;
this.first_byte = true;
this._who_am_i_count = 0;
this.regs = new Uint8Array(256);
// WHO_AM_I = 0x68
this.regs[0x75] = 0x68;
// ACCEL_Z = +1g = 0x40 (MSB at 0x3F)
this.regs[0x3F] = 0x40;
// TEMP raw for 25°C
const tempRaw = Math.round((25.0 - 36.53) * 340) & 0xFFFF;
this.regs[0x41] = (tempRaw >> 8) & 0xFF;
this.regs[0x42] = tempRaw & 0xFF;
this.eventCount = 0;
this.readCount = 0;
}
handle(event) {
const op = event & 0xFF;
const data = (event >> 8) & 0xFF;
this.eventCount++;
let val = 0;
let note = '';
if (op === 0x01) { // I2C_START
this.first_byte = true;
if (this._who_am_i_count >= 2) {
this.reg_ptr = 0x3B;
note = `DATA_MODE (count=${this._who_am_i_count} ≥ 2)`;
} else {
this.reg_ptr = 0x75;
note = `WHO_AM_I_MODE (count=${this._who_am_i_count} < 2)`;
}
val = 1;
} else if (op === 0x05 || op === 0x06) { // WRITE
if (this.first_byte) {
this.reg_ptr = data;
this.first_byte = false;
note = `set reg_ptr=0x${data.toString(16).padStart(2,'0')} (${REG_NAME[data] ?? '?'})`;
} else {
this.regs[this.reg_ptr] = data;
if (this.reg_ptr === 0x6B) this.regs[0x6B] &= 0x7F; // auto-clear DEVICE_RESET
note = `write 0x${data.toString(16).padStart(2,'0')} → reg[0x${this.reg_ptr.toString(16).padStart(2,'0')}]`;
this.reg_ptr = (this.reg_ptr + 1) & 0xFF;
}
val = 1;
} else if (op === 0x03) { // READ
this.readCount++;
val = this.regs[this.reg_ptr];
if (this.reg_ptr === 0x75 && val === 0x68) {
this._who_am_i_count++;
note = `WHO_AM_I read #${this._who_am_i_count} → 0x68`;
} else {
note = `reg[0x${this.reg_ptr.toString(16).padStart(2,'0')}]=${FRIENDLY_REG(this.reg_ptr, val)}`;
}
this.reg_ptr = (this.reg_ptr + 1) & 0xFF;
} else { // STOP
this.first_byte = true;
note = 'transaction end';
}
return { val, note };
}
/** Guess which firmware function triggered this event sequence */
guessPhase() {
const c = this._who_am_i_count;
const r = this.readCount;
if (c === 0) return 'before_begin / detected()';
if (c === 1) return 'chip_id.read() or reset() read-modify-write';
if (c === 2 && r <= 3) return 'reset() wait loop or setFilterBandwidth';
if (c >= 2) return '_init() config setup OR sketch getEvent() loop';
return '?';
}
}
// Known MPU-6050 register names
const REG_NAME = {
0x19: 'SMPRT_DIV', 0x1A: 'CONFIG', 0x1B: 'GYRO_CONFIG', 0x1C: 'ACCEL_CONFIG',
0x3B: 'ACCEL_XOUT_H', 0x3C: 'ACCEL_XOUT_L', 0x3D: 'ACCEL_YOUT_H', 0x3E: 'ACCEL_YOUT_L',
0x3F: 'ACCEL_ZOUT_H', 0x40: 'ACCEL_ZOUT_L',
0x41: 'TEMP_OUT_H', 0x42: 'TEMP_OUT_L',
0x43: 'GYRO_XOUT_H', 0x44: 'GYRO_XOUT_L', 0x45: 'GYRO_YOUT_H', 0x46: 'GYRO_YOUT_L',
0x47: 'GYRO_ZOUT_H', 0x48: 'GYRO_ZOUT_L',
0x6B: 'PWR_MGMT_1', 0x6C: 'PWR_MGMT_2',
0x68: 'SIGNAL_PATH_RESET',
0x75: 'WHO_AM_I',
};
function FRIENDLY_REG(reg, val) {
const name = REG_NAME[reg];
const hex = `0x${val.toString(16).padStart(2,'0')}`;
if (name) return `${hex} [${name}]`;
return hex;
}
// ─── Logging helpers ───────────────────────────────────────────────────────────
const T0 = Date.now();
const ts = () => `[+${((Date.now() - T0)/1000).toFixed(3)}s]`;
const LOG_LEVELS = { INFO: '\x1b[36m', WARN: '\x1b[33m', ERROR: '\x1b[31m', OK: '\x1b[32m', I2C: '\x1b[35m', SERIAL: '\x1b[32m', RESET: '\x1b[0m' };
const log = (lvl, ...args) => console.log(`${LOG_LEVELS[lvl] ?? ''}${ts()} [${lvl}]${LOG_LEVELS.RESET}`, ...args);
const info = (...a) => log('INFO', ...a);
const warn = (...a) => log('WARN', ...a);
const ok = (...a) => log('OK', ...a);
const err = (...a) => log('ERROR', ...a);
const i2c = (...a) => log('I2C', ...a);
const serial = (...a) => log('SERIAL', ...a);
// ─── Counters & state ─────────────────────────────────────────────────────────
let totalEvents = 0;
let i2cEvents = 0;
let i2cTraceCount = 0;
let serialLines = [];
let foundOK = false;
let foundFail = false;
const mirror = new MPU6050StateMirror();
// ─── Step 1: Compile the sketch ───────────────────────────────────────────────
async function compile() {
info('Compiling MPU6050 sketch via POST /api/compile/ ...');
const res = await fetch(`${BACKEND}/api/compile/`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
files: [{ name: 'sketch.ino', content: SKETCH }],
board_fqbn: 'esp32:esp32:esp32',
}),
});
if (!res.ok) {
const text = await res.text();
throw new Error(`Compilation failed HTTP ${res.status}: ${text.slice(0, 300)}`);
}
const body = await res.json();
if (!body.success) {
throw new Error(`Compilation error:\n${(body.error ?? body.stderr ?? 'unknown').slice(0, 500)}`);
}
// API returns firmware as base64 in 'binary_content' (ESP32 flash image)
const firmware_b64 = body.binary_content ?? body.firmware_b64;
if (!firmware_b64) {
throw new Error(`No firmware in response. Keys: ${Object.keys(body).join(', ')}`);
}
const sizeKB = Math.round(firmware_b64.length * 0.75 / 1024);
ok(`Compilation succeeded — ${sizeKB} KB firmware (has_wifi=${body.has_wifi})`);
return firmware_b64;
}
// ─── Step 2: Run simulation via WebSocket ─────────────────────────────────────
function runSimulation(firmware_b64) {
return new Promise((resolve, reject) => {
const wsUrl = `${WS_BASE}/api/simulation/ws/${SESSION}`;
info(`Connecting WebSocket → ${wsUrl}`);
const ws = new WebSocket(wsUrl);
const timer = setTimeout(() => {
info(`Timeout reached (${TIMEOUT_S}s) — stopping simulation`);
ws.close();
resolve({ timedOut: true });
}, TIMEOUT_S * 1000);
ws.addEventListener('open', () => {
ok('WebSocket connected');
const payload = {
type: 'start_esp32',
data: {
board: 'esp32',
firmware_b64,
sensors: [
// Mirror what useSimulatorStore sends for wokwi-mpu6050 component
{ sensor_type: 'mpu6050', pin: 200 + 0x68, addr: 0x68 }
],
wifi_enabled: false,
},
};
info('Sending start_esp32 with sensors:', JSON.stringify(payload.data.sensors));
ws.send(JSON.stringify(payload));
});
ws.addEventListener('message', ev => {
totalEvents++;
let msg;
try { msg = JSON.parse(ev.data); } catch { return; }
const { type, data } = msg;
// ── Serial output ──────────────────────────────────────────────────────
if (type === 'serial_output') {
const text = (data?.data ?? '').trim();
if (!text) return;
for (const line of text.split(/\r?\n/)) {
if (!line.trim()) continue;
serialLines.push(line);
serial(`UART: ${line}`);
if (line.includes('MPU6050 ready!') || line.includes('===BEGIN_OK===')) {
foundOK = true;
}
if (line.includes('not found') || line.includes('===BEGIN_FAILED===')) {
foundFail = true;
warn('begin() returned FALSE — firmware reported "not found"');
}
// If we got sensor data, stop after a few lines
if (foundOK && serialLines.filter(l => l.startsWith('Accel')).length >= 3) {
clearTimeout(timer);
ws.close();
resolve({ timedOut: false });
}
}
// Fail fast: if error already confirmed, still wait for a bit more events
if (foundFail && i2cEvents > 5) {
// Give 5 more seconds to see remaining I2C events
setTimeout(() => { clearTimeout(timer); ws.close(); resolve({ timedOut: false }); }, 5000);
}
return;
}
// ── I2C trace (Python slave handled the event — always emitted for debug) ─
if (type === 'i2c_trace') {
i2cTraceCount++;
const { bus, addr, event, op, result, reg_ptr, wai_count } = data;
const regHex = reg_ptr != null ? `0x${reg_ptr.toString(16).padStart(2,'0')}` : '??';
const resHex = `0x${(result??0).toString(16).padStart(2,'0')}`;
i2c(`[slave] bus=${bus} addr=0x${(addr??0).toString(16).padStart(2,'0')} ` +
`op=${op} result=${resHex} reg_ptr=${regHex} wai=${wai_count}`);
return;
}
// ── I2C event (forwarded from Python worker when addr NOT in _i2c_slaves) ─
if (type === 'i2c_event') {
i2cEvents++;
const { bus, addr, event, response } = data;
const op = event & 0xFF;
const d = (event >> 8) & 0xFF;
const opName = I2C_OP[op] ?? `0x${op.toString(16)}`;
i2c(`[UNHANDLED slave] bus=${bus} addr=0x${(addr??0).toString(16).padStart(2,'0')} ` +
`event=0x${(event??0).toString(16).padStart(4,'0')} op=${opName} data=0x${d.toString(16).padStart(2,'0')} ` +
`resp=0x${(response??0).toString(16).padStart(2,'0')}`);
return;
}
// ── gpio_change — show I2C pin activity ───────────────────────────────
if (type === 'gpio_change') {
const { pin, state } = data ?? {};
if (pin === 21 || pin === 22) {
// SDA=21, SCL=22 — these toggling means I2C is active on the bus
i2c(`I2C pin toggle: GPIO${pin} (${pin===21?'SDA':'SCL'}) → ${state}`);
}
return;
}
// ── system / error ─────────────────────────────────────────────────────
if (type === 'system') {
info(`system event: ${JSON.stringify(data)}`);
return;
}
if (type === 'error') {
err(`simulation error: ${JSON.stringify(data)}`);
return;
}
// ── Everything else ───────────────────────────────────────────────────
if (!['gpio_change'].includes(type)) {
info(`event type=${type} data=${JSON.stringify(data).slice(0,120)}`);
}
});
ws.addEventListener('close', ev => {
clearTimeout(timer);
info(`WebSocket closed (code=${ev.code})`);
resolve({ timedOut: false });
});
ws.addEventListener('error', ev => {
clearTimeout(timer);
err('WebSocket error:', ev.message ?? ev.type);
reject(new Error('WebSocket error'));
});
});
}
// ─── Main ──────────────────────────────────────────────────────────────────────
async function main() {
console.log('\n' + '═'.repeat(60));
console.log(' TEST: ESP32 + MPU-6050 I2C Simulation');
console.log(' Session:', SESSION);
console.log(' Backend:', BACKEND);
console.log(' Timeout:', TIMEOUT_S, 's');
console.log('═'.repeat(60) + '\n');
let firmware_b64;
try {
firmware_b64 = await compile();
} catch (e) {
err('Compilation failed:', e.message);
process.exit(1);
}
console.log('\n' + '─'.repeat(60));
console.log(' Starting simulation...');
console.log('─'.repeat(60) + '\n');
info('NOTE: I2C trace events emitted by backend are shown below with [slave] prefix.');
console.log();
const result = await runSimulation(firmware_b64);
// ─── Summary ────────────────────────────────────────────────────────────────
console.log('\n' + '═'.repeat(60));
console.log(' SUMMARY');
console.log('═'.repeat(60));
console.log(` Total WebSocket events received : ${totalEvents}`);
console.log(` I2C trace events (slave handled): ${i2cTraceCount}`);
console.log(` I2C events (no slave registered): ${i2cEvents}`);
console.log(` Serial lines received : ${serialLines.length}`);
console.log(` Timed out : ${result.timedOut}`);
console.log();
console.log(' Serial output:');
for (const l of serialLines) console.log(` ${l}`);
console.log();
if (foundOK) {
console.log('\x1b[32m ✓ PASS — MPU6050 detected and sensor data flowing\x1b[0m');
process.exit(0);
} else if (foundFail) {
console.log('\x1b[31m ✗ FAIL — mpu.begin() returned false ("not found")\x1b[0m');
console.log('\x1b[33m → Check the backend (uvicorn) terminal for I2C event trace.\x1b[0m');
console.log('\x1b[33m → Look for "I2C bus=0 addr=0x68" lines to see the full sequence.\x1b[0m');
process.exit(1);
} else if (result.timedOut) {
console.log('\x1b[33m ? TIMEOUT — no "ready" or "not found" in serial output\x1b[0m');
process.exit(1);
} else {
console.log('\x1b[33m ? INCONCLUSIVE — WebSocket closed before result determined\x1b[0m');
process.exit(1);
}
}
main().catch(e => { err('Unhandled error:', e); process.exit(1); });

2
frontend/public/components-metadata.json
View File

@ -1,6 +1,6 @@
{
"version": "1.0.0",
"generatedAt": "2026-04-08T23:38:55.702Z",
"generatedAt": "2026-04-09T12:43:34.605Z",
"components": [
{
"id": "arduino-mega",