646 lines
27 KiB
Python
646 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
esp32_worker.py — Standalone ESP32 QEMU subprocess worker.
|
|
|
|
Runs as a child process of esp32_lib_manager. Loads libqemu-xtensa in its
|
|
own process address space so multiple instances can coexist without DLL state
|
|
conflicts.
|
|
|
|
stdin line 1 : JSON config
|
|
{"lib_path": "...", "firmware_b64": "...", "machine": "..."}
|
|
stdin line 2+: JSON commands
|
|
{"cmd": "set_pin", "pin": N, "value": V}
|
|
{"cmd": "set_adc", "channel": N, "millivolts": V}
|
|
{"cmd": "set_adc_raw", "channel": N, "raw": V}
|
|
{"cmd": "uart_send", "uart": N, "data": "<base64>"}
|
|
{"cmd": "set_i2c_response", "addr": N, "response": V}
|
|
{"cmd": "set_spi_response", "response": V}
|
|
{"cmd": "stop"}
|
|
|
|
stdout : JSON event lines (one per line, flushed immediately)
|
|
{"type": "system", "event": "booted"}
|
|
{"type": "system", "event": "crash", "reason": "...", ...}
|
|
{"type": "system", "event": "reboot", "count": N}
|
|
{"type": "gpio_change", "pin": N, "state": V}
|
|
{"type": "gpio_dir", "pin": N, "dir": V}
|
|
{"type": "uart_tx", "uart": N, "byte": V}
|
|
{"type": "ledc_update", "channel": N, "duty": V, "duty_pct": F, "gpio": N|-1}
|
|
{"type": "rmt_event", "channel": N, ...}
|
|
{"type": "ws2812_update","channel": N, "pixels": [...]}
|
|
{"type": "i2c_event", "bus": N, "addr": N, "event": N, "response": N}
|
|
{"type": "spi_event", "bus": N, "event": N, "response": N}
|
|
{"type": "error", "message": "..."}
|
|
|
|
stderr : debug logs (never part of the JSON protocol)
|
|
"""
|
|
import base64
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
# ─── stdout helpers ──────────────────────────────────────────────────────────
|
|
|
|
_stdout_lock = threading.Lock()
|
|
|
|
|
|
def _emit(obj: dict) -> None:
|
|
"""Write one JSON event line to stdout (thread-safe, always flushed)."""
|
|
with _stdout_lock:
|
|
sys.stdout.write(json.dumps(obj) + '\n')
|
|
sys.stdout.flush()
|
|
|
|
|
|
def _log(msg: str) -> None:
|
|
"""Write a debug message to stderr (invisible to parent's stdout reader)."""
|
|
sys.stderr.write(f'[esp32_worker] {msg}\n')
|
|
sys.stderr.flush()
|
|
|
|
|
|
# ─── GPIO pinmap (identity: slot i → GPIO i-1) ──────────────────────────────
|
|
# ESP32 has 40 GPIOs (0-39), ESP32-C3 only has 22 (0-21).
|
|
# The pinmap is rebuilt after reading config (see main()), defaulting to ESP32.
|
|
|
|
_GPIO_COUNT = 40
|
|
_PINMAP = (ctypes.c_int16 * (_GPIO_COUNT + 1))(
|
|
_GPIO_COUNT,
|
|
*range(_GPIO_COUNT),
|
|
)
|
|
|
|
|
|
def _build_pinmap(gpio_count: int):
|
|
"""Build a pinmap array for the given GPIO count."""
|
|
global _GPIO_COUNT, _PINMAP
|
|
_GPIO_COUNT = gpio_count
|
|
_PINMAP = (ctypes.c_int16 * (gpio_count + 1))(
|
|
gpio_count,
|
|
*range(gpio_count),
|
|
)
|
|
|
|
# ─── ctypes callback types ───────────────────────────────────────────────────
|
|
|
|
_WRITE_PIN = ctypes.CFUNCTYPE(None, ctypes.c_int, ctypes.c_int)
|
|
_DIR_PIN = ctypes.CFUNCTYPE(None, ctypes.c_int, ctypes.c_int)
|
|
_I2C_EVENT = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint16)
|
|
_SPI_EVENT = ctypes.CFUNCTYPE(ctypes.c_uint8, ctypes.c_uint8, ctypes.c_uint16)
|
|
_UART_TX = ctypes.CFUNCTYPE(None, ctypes.c_uint8, ctypes.c_uint8)
|
|
_RMT_EVENT = ctypes.CFUNCTYPE(None, ctypes.c_uint8, ctypes.c_uint32, ctypes.c_uint32)
|
|
|
|
|
|
class _CallbacksT(ctypes.Structure):
|
|
_fields_ = [
|
|
('picsimlab_write_pin', _WRITE_PIN),
|
|
('picsimlab_dir_pin', _DIR_PIN),
|
|
('picsimlab_i2c_event', _I2C_EVENT),
|
|
('picsimlab_spi_event', _SPI_EVENT),
|
|
('picsimlab_uart_tx_event', _UART_TX),
|
|
('pinmap', ctypes.c_void_p),
|
|
('picsimlab_rmt_event', _RMT_EVENT),
|
|
]
|
|
|
|
|
|
# ─── RMT / WS2812 NeoPixel decoder ───────────────────────────────────────────
|
|
|
|
_WS2812_HIGH_THRESHOLD = 48 # RMT ticks; high pulse > threshold → bit 1
|
|
|
|
|
|
def _decode_rmt_item(value: int) -> tuple[int, int, int, int]:
|
|
"""Unpack a 32-bit RMT item → (level0, duration0, level1, duration1)."""
|
|
level0 = (value >> 31) & 1
|
|
duration0 = (value >> 16) & 0x7FFF
|
|
level1 = (value >> 15) & 1
|
|
duration1 = value & 0x7FFF
|
|
return level0, duration0, level1, duration1
|
|
|
|
|
|
class _RmtDecoder:
|
|
"""Accumulate RMT items for one channel; flush complete WS2812 frames."""
|
|
|
|
def __init__(self, channel: int):
|
|
self.channel = channel
|
|
self._bits: list[int] = []
|
|
self._pixels: list[dict] = []
|
|
|
|
@staticmethod
|
|
def _bits_to_byte(bits: list[int], offset: int) -> int:
|
|
val = 0
|
|
for i in range(8):
|
|
val = (val << 1) | bits[offset + i]
|
|
return val
|
|
|
|
def feed(self, value: int) -> list[dict] | None:
|
|
"""
|
|
Process one RMT item.
|
|
Returns a list of {r, g, b} pixel dicts on end-of-frame, else None.
|
|
"""
|
|
level0, dur0, _, dur1 = _decode_rmt_item(value)
|
|
|
|
# Reset pulse (both durations zero) signals end of frame
|
|
if dur0 == 0 and dur1 == 0:
|
|
pix = list(self._pixels)
|
|
self._pixels.clear()
|
|
self._bits.clear()
|
|
return pix or None
|
|
|
|
# Classify the high pulse → bit 1 or bit 0
|
|
if level0 == 1 and dur0 > 0:
|
|
self._bits.append(1 if dur0 > _WS2812_HIGH_THRESHOLD else 0)
|
|
|
|
# Every 24 bits → one GRB pixel → convert to RGB
|
|
while len(self._bits) >= 24:
|
|
g = self._bits_to_byte(self._bits, 0)
|
|
r = self._bits_to_byte(self._bits, 8)
|
|
b = self._bits_to_byte(self._bits, 16)
|
|
self._pixels.append({'r': r, 'g': g, 'b': b})
|
|
self._bits = self._bits[24:]
|
|
|
|
return None
|
|
|
|
|
|
# ─── Main ─────────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None: # noqa: C901 (complexity OK for inline worker)
|
|
# ── 1. Read config from stdin ─────────────────────────────────────────────
|
|
raw_cfg = sys.stdin.readline()
|
|
if not raw_cfg.strip():
|
|
_log('No config received on stdin — exiting')
|
|
os._exit(1)
|
|
try:
|
|
cfg = json.loads(raw_cfg)
|
|
except Exception as exc:
|
|
_log(f'Bad config JSON: {exc}')
|
|
os._exit(1)
|
|
|
|
lib_path = cfg['lib_path']
|
|
firmware_b64 = cfg['firmware_b64']
|
|
machine = cfg.get('machine', 'esp32-picsimlab')
|
|
initial_sensors = cfg.get('sensors', [])
|
|
|
|
# Adjust GPIO pinmap based on chip: ESP32-C3 has only 22 GPIOs
|
|
if 'c3' in machine:
|
|
_build_pinmap(22)
|
|
|
|
# ── 2. Load DLL ───────────────────────────────────────────────────────────
|
|
_MINGW64_BIN = r'C:\msys64\mingw64\bin'
|
|
if os.name == 'nt' and os.path.isdir(_MINGW64_BIN):
|
|
os.add_dll_directory(_MINGW64_BIN)
|
|
try:
|
|
lib = ctypes.CDLL(lib_path)
|
|
except Exception as exc:
|
|
_emit({'type': 'error', 'message': f'Cannot load DLL: {exc}'})
|
|
os._exit(1)
|
|
|
|
# ── 3. Write firmware to a temp file ──────────────────────────────────────
|
|
try:
|
|
fw_bytes = base64.b64decode(firmware_b64)
|
|
tmp = tempfile.NamedTemporaryFile(suffix='.bin', delete=False)
|
|
tmp.write(fw_bytes)
|
|
tmp.close()
|
|
firmware_path: str | None = tmp.name
|
|
except Exception as exc:
|
|
_emit({'type': 'error', 'message': f'Firmware decode error: {exc}'})
|
|
os._exit(1)
|
|
|
|
rom_dir = os.path.dirname(lib_path).encode()
|
|
args_list = [
|
|
b'qemu',
|
|
b'-M', machine.encode(),
|
|
b'-nographic',
|
|
b'-L', rom_dir,
|
|
b'-drive', f'file={firmware_path},if=mtd,format=raw'.encode(),
|
|
]
|
|
argc = len(args_list)
|
|
argv = (ctypes.c_char_p * argc)(*args_list)
|
|
|
|
# ── 4. Shared mutable state ───────────────────────────────────────────────
|
|
_stopped = threading.Event() # set on "stop" command
|
|
_init_done = threading.Event() # set when qemu_init() returns
|
|
_sensors_ready = threading.Event() # set after pre-registering initial sensors
|
|
_i2c_responses: dict[int, int] = {} # 7-bit addr → response byte
|
|
_spi_response = [0xFF] # MISO byte for SPI transfers
|
|
_rmt_decoders: dict[int, _RmtDecoder] = {}
|
|
_uart0_buf = bytearray() # accumulate UART0 for crash detection
|
|
_reboot_count = [0]
|
|
_crashed = [False]
|
|
_CRASH_STR = b'Cache disabled but cached memory region accessed'
|
|
_REBOOT_STR = b'Rebooting...'
|
|
# LEDC channel → GPIO pin (populated from GPIO out_sel sync events)
|
|
# ESP32 signal indices: 72-79 = LEDC HS ch 0-7, 80-87 = LEDC LS ch 0-7
|
|
_ledc_gpio_map: dict[int, int] = {}
|
|
|
|
# Sensor state: gpio_pin → {type, properties..., saw_low, responding}
|
|
_sensors: dict[int, dict] = {}
|
|
_sensors_lock = threading.Lock()
|
|
|
|
def _busy_wait_us(us: int) -> None:
|
|
"""Busy-wait for the given number of microseconds using perf_counter_ns."""
|
|
end = time.perf_counter_ns() + us * 1000
|
|
while time.perf_counter_ns() < end:
|
|
pass
|
|
|
|
def _dht22_build_payload(temperature: float, humidity: float) -> list[int]:
|
|
"""Build 5-byte DHT22 data payload: [hum_H, hum_L, temp_H, temp_L, checksum]."""
|
|
hum = round(humidity * 10)
|
|
tmp = round(temperature * 10)
|
|
h_H = (hum >> 8) & 0xFF
|
|
h_L = hum & 0xFF
|
|
raw_t = ((-tmp) & 0x7FFF) | 0x8000 if tmp < 0 else tmp & 0x7FFF
|
|
t_H = (raw_t >> 8) & 0xFF
|
|
t_L = raw_t & 0xFF
|
|
chk = (h_H + h_L + t_H + t_L) & 0xFF
|
|
return [h_H, h_L, t_H, t_L, chk]
|
|
|
|
def _dht22_respond(gpio_pin: int, temperature: float, humidity: float,
|
|
ratio: float) -> None:
|
|
"""Thread function: inject the DHT22 protocol waveform via qemu_picsimlab_set_pin.
|
|
|
|
The pin is already driven LOW by the _on_dir_change callback before this
|
|
thread starts. *ratio* is wall-clock-µs per QEMU-µs, derived from
|
|
measuring how long the firmware's start signal took in wall-clock time.
|
|
"""
|
|
slot = gpio_pin + 1 # identity pinmap: slot = gpio + 1
|
|
payload = _dht22_build_payload(temperature, humidity)
|
|
|
|
def qemu_wait(qemu_us: float) -> None:
|
|
"""Busy-wait for the wall-clock equivalent of *qemu_us* QEMU µs."""
|
|
wall_us = max(1, int(qemu_us * ratio))
|
|
_busy_wait_us(wall_us)
|
|
|
|
try:
|
|
_log(f'DHT22 respond: gpio={gpio_pin} slot={slot} ratio={ratio:.4f}')
|
|
|
|
# Pin is already LOW (driven synchronously in _on_dir_change).
|
|
# Preamble: hold LOW 80 µs → drive HIGH 80 µs
|
|
qemu_wait(80)
|
|
lib.qemu_picsimlab_set_pin(slot, 1)
|
|
qemu_wait(80)
|
|
|
|
# 40 data bits: 50 µs LOW + (26 µs HIGH = 0, 70 µs HIGH = 1)
|
|
for byte_val in payload:
|
|
for b in range(7, -1, -1):
|
|
bit = (byte_val >> b) & 1
|
|
lib.qemu_picsimlab_set_pin(slot, 0)
|
|
qemu_wait(50)
|
|
lib.qemu_picsimlab_set_pin(slot, 1)
|
|
qemu_wait(70 if bit else 26)
|
|
|
|
# Final: release line HIGH
|
|
lib.qemu_picsimlab_set_pin(slot, 0)
|
|
qemu_wait(50)
|
|
lib.qemu_picsimlab_set_pin(slot, 1)
|
|
except Exception as exc:
|
|
_log(f'DHT22 respond error on GPIO {gpio_pin}: {exc}')
|
|
finally:
|
|
_log(f'DHT22 respond done on GPIO {gpio_pin}')
|
|
with _sensors_lock:
|
|
sensor = _sensors.get(gpio_pin)
|
|
if sensor:
|
|
sensor['responding'] = False
|
|
|
|
def _hcsr04_respond(trig_pin: int, echo_pin: int, distance_cm: float) -> None:
|
|
"""Thread function: inject the HC-SR04 echo pulse via qemu_picsimlab_set_pin."""
|
|
echo_slot = echo_pin + 1 # identity pinmap: slot = gpio + 1
|
|
# Echo pulse width = distance_cm * 58 µs (speed of sound round trip)
|
|
echo_us = max(100, int(distance_cm * 58))
|
|
|
|
try:
|
|
# Wait for TRIG pulse to finish + propagation delay (~600 µs)
|
|
_busy_wait_us(600)
|
|
# Drive ECHO HIGH
|
|
lib.qemu_picsimlab_set_pin(echo_slot, 1)
|
|
# Hold ECHO HIGH for distance-proportional duration
|
|
_busy_wait_us(echo_us)
|
|
# Drive ECHO LOW
|
|
lib.qemu_picsimlab_set_pin(echo_slot, 0)
|
|
except Exception as exc:
|
|
_log(f'HC-SR04 respond error on TRIG {trig_pin} ECHO {echo_pin}: {exc}')
|
|
finally:
|
|
with _sensors_lock:
|
|
sensor = _sensors.get(trig_pin)
|
|
if sensor:
|
|
sensor['responding'] = False
|
|
|
|
# ── 5. ctypes callbacks (called from QEMU thread) ─────────────────────────
|
|
|
|
def _on_pin_change(slot: int, value: int) -> None:
|
|
if _stopped.is_set():
|
|
return
|
|
gpio = int(_PINMAP[slot]) if 1 <= slot <= _GPIO_COUNT else slot
|
|
_emit({'type': 'gpio_change', 'pin': gpio, 'state': value})
|
|
|
|
# Sensor protocol dispatch by type
|
|
with _sensors_lock:
|
|
sensor = _sensors.get(gpio)
|
|
if sensor is None:
|
|
return
|
|
|
|
stype = sensor.get('type', '')
|
|
|
|
if stype == 'dht22':
|
|
# Record that the firmware drove the pin LOW (start signal).
|
|
# The actual response is triggered from _on_dir_change when the
|
|
# firmware switches the pin to INPUT mode.
|
|
if value == 0 and not sensor.get('responding', False):
|
|
sensor['saw_low'] = True
|
|
|
|
elif stype == 'hc-sr04':
|
|
# HC-SR04: detect TRIG going HIGH (firmware sends 10µs pulse)
|
|
if value == 1 and not sensor.get('responding', False):
|
|
sensor['responding'] = True
|
|
echo_pin = int(sensor.get('echo_pin', gpio + 1))
|
|
distance = float(sensor.get('distance', 40.0))
|
|
threading.Thread(
|
|
target=_hcsr04_respond,
|
|
args=(gpio, echo_pin, distance),
|
|
daemon=True,
|
|
name=f'hcsr04-gpio{gpio}',
|
|
).start()
|
|
|
|
def _on_dir_change(slot: int, direction: int) -> None:
|
|
if _stopped.is_set():
|
|
return
|
|
|
|
# Debug: log all real-pin direction changes (skip noisy slot=-1 sync events)
|
|
if slot >= 1:
|
|
gpio_dbg = int(_PINMAP[slot]) if slot <= _GPIO_COUNT else slot
|
|
_log(f'DIR_CHANGE slot={slot} gpio={gpio_dbg} direction={direction}')
|
|
|
|
# DHT22: track direction changes for calibration + response trigger.
|
|
# QEMU runs faster than real-time, so wall-clock _busy_wait_us()
|
|
# delays are too slow. We calibrate by measuring the wall-clock
|
|
# duration of the firmware's start signal (OUTPUT→INPUT), which
|
|
# corresponds to a known QEMU duration (~1200 µs minimum).
|
|
if slot >= 1:
|
|
gpio = int(_PINMAP[slot]) if slot <= _GPIO_COUNT else slot
|
|
with _sensors_lock:
|
|
sensor = _sensors.get(gpio)
|
|
if sensor is not None and sensor.get('type') == 'dht22':
|
|
if direction == 1:
|
|
# OUTPUT mode — record timestamp for timing calibration
|
|
sensor['dir_out_ns'] = time.perf_counter_ns()
|
|
elif direction == 0:
|
|
# INPUT mode — trigger DHT22 response
|
|
if sensor.get('saw_low', False) and not sensor.get('responding', False):
|
|
sensor['saw_low'] = False
|
|
sensor['responding'] = True
|
|
|
|
# Calibrate: measure how long the start signal took in
|
|
# wall-clock time. The QEMU time between direction=1
|
|
# and direction=0 is at least ~1200 µs (Adafruit DHT
|
|
# library: delayMicroseconds(1100) + overhead).
|
|
now_ns = time.perf_counter_ns()
|
|
dir_out_ns = sensor.get('dir_out_ns', now_ns)
|
|
wall_us = max(1.0, (now_ns - dir_out_ns) / 1000)
|
|
qemu_us_signal = 1200.0
|
|
ratio = wall_us / qemu_us_signal
|
|
_log(f'DHT22 dir_change→INPUT gpio={gpio}: '
|
|
f'wall={wall_us:.0f}µs ratio={ratio:.4f}')
|
|
|
|
# Drive pin LOW *synchronously* before returning to
|
|
# QEMU — this guarantees the firmware sees LOW at its
|
|
# first digitalRead() in expectPulse().
|
|
lib.qemu_picsimlab_set_pin(slot, 0)
|
|
|
|
threading.Thread(
|
|
target=_dht22_respond,
|
|
args=(gpio, sensor.get('temperature', 25.0),
|
|
sensor.get('humidity', 50.0), ratio),
|
|
daemon=True,
|
|
name=f'dht22-gpio{gpio}',
|
|
).start()
|
|
|
|
# slot == -1 means a sync event from GPIO/LEDC/IOMUX peripheral
|
|
if slot == -1:
|
|
marker = direction & 0xF000
|
|
if marker == 0x2000: # GPIO_FUNCX_OUT_SEL_CFG change
|
|
gpio_pin = direction & 0xFF
|
|
signal = (direction >> 8) & 0xFF
|
|
# Signal 72-79 = LEDC HS ch 0-7; 80-87 = LEDC LS ch 0-7
|
|
if 72 <= signal <= 87:
|
|
ledc_ch = signal - 72 # ch 0-15
|
|
_ledc_gpio_map[ledc_ch] = gpio_pin
|
|
return
|
|
gpio = int(_PINMAP[slot]) if 1 <= slot <= _GPIO_COUNT else slot
|
|
_emit({'type': 'gpio_dir', 'pin': gpio, 'dir': direction})
|
|
|
|
def _on_uart_tx(uart_id: int, byte_val: int) -> None:
|
|
if _stopped.is_set():
|
|
return
|
|
_emit({'type': 'uart_tx', 'uart': uart_id, 'byte': byte_val})
|
|
# Crash / reboot detection on UART0 only
|
|
if uart_id == 0:
|
|
_uart0_buf.append(byte_val)
|
|
if byte_val == ord('\n') or len(_uart0_buf) >= 512:
|
|
chunk = bytes(_uart0_buf)
|
|
_uart0_buf.clear()
|
|
if _CRASH_STR in chunk and not _crashed[0]:
|
|
_crashed[0] = True
|
|
_emit({'type': 'system', 'event': 'crash',
|
|
'reason': 'cache_error', 'reboot': _reboot_count[0]})
|
|
if _REBOOT_STR in chunk:
|
|
_crashed[0] = False
|
|
_reboot_count[0] += 1
|
|
_emit({'type': 'system', 'event': 'reboot',
|
|
'count': _reboot_count[0]})
|
|
|
|
def _on_rmt_event(channel: int, config0: int, value: int) -> None:
|
|
if _stopped.is_set():
|
|
return
|
|
level0, dur0, level1, dur1 = _decode_rmt_item(value)
|
|
_emit({'type': 'rmt_event', 'channel': channel, 'config0': config0,
|
|
'value': value, 'level0': level0, 'dur0': dur0,
|
|
'level1': level1, 'dur1': dur1})
|
|
if channel not in _rmt_decoders:
|
|
_rmt_decoders[channel] = _RmtDecoder(channel)
|
|
pixels = _rmt_decoders[channel].feed(value)
|
|
if pixels:
|
|
_emit({'type': 'ws2812_update', 'channel': channel, 'pixels': pixels})
|
|
|
|
def _on_i2c_event(bus_id: int, addr: int, event: int) -> int:
|
|
"""Synchronous — must return immediately; called from QEMU thread."""
|
|
resp = _i2c_responses.get(addr, 0)
|
|
if not _stopped.is_set():
|
|
_emit({'type': 'i2c_event', 'bus': bus_id, 'addr': addr,
|
|
'event': event, 'response': resp})
|
|
return resp
|
|
|
|
def _on_spi_event(bus_id: int, event: int) -> int:
|
|
"""Synchronous — must return immediately; called from QEMU thread."""
|
|
resp = _spi_response[0]
|
|
if not _stopped.is_set():
|
|
_emit({'type': 'spi_event', 'bus': bus_id, 'event': event, 'response': resp})
|
|
return resp
|
|
|
|
# Keep callback struct alive (prevent GC from freeing ctypes closures)
|
|
_cbs_ref = _CallbacksT(
|
|
picsimlab_write_pin = _WRITE_PIN(_on_pin_change),
|
|
picsimlab_dir_pin = _DIR_PIN(_on_dir_change),
|
|
picsimlab_i2c_event = _I2C_EVENT(_on_i2c_event),
|
|
picsimlab_spi_event = _SPI_EVENT(_on_spi_event),
|
|
picsimlab_uart_tx_event = _UART_TX(_on_uart_tx),
|
|
pinmap = ctypes.cast(_PINMAP, ctypes.c_void_p).value,
|
|
picsimlab_rmt_event = _RMT_EVENT(_on_rmt_event),
|
|
)
|
|
lib.qemu_picsimlab_register_callbacks(ctypes.byref(_cbs_ref))
|
|
|
|
# ── 6. QEMU thread ────────────────────────────────────────────────────────
|
|
|
|
def _qemu_thread() -> None:
|
|
try:
|
|
lib.qemu_init(argc, argv, None)
|
|
except Exception as exc:
|
|
_emit({'type': 'error', 'message': f'qemu_init failed: {exc}'})
|
|
finally:
|
|
_init_done.set()
|
|
# Wait for initial sensors to be pre-registered before executing firmware.
|
|
# This prevents race conditions where the firmware tries to read a sensor
|
|
# (e.g. DHT22 pulseIn) before the sensor handler is registered.
|
|
_sensors_ready.wait(timeout=5.0)
|
|
lib.qemu_main_loop()
|
|
|
|
# With -nographic, qemu_init registers the stdio mux chardev which reads
|
|
# from fd 0. If we leave fd 0 as the JSON-command pipe from the parent,
|
|
# QEMU's mux will consume those bytes and forward them to UART0 RX,
|
|
# corrupting user-sent serial data. Redirect fd 0 to /dev/null before
|
|
# qemu_init runs so the mux gets EOF and leaves our command pipe alone.
|
|
# Save the original pipe fd for the command loop below.
|
|
_orig_stdin_fd = os.dup(0)
|
|
_nul = os.open(os.devnull, os.O_RDONLY)
|
|
os.dup2(_nul, 0)
|
|
os.close(_nul)
|
|
|
|
qemu_t = threading.Thread(target=_qemu_thread, daemon=True, name=f'qemu-{machine}')
|
|
qemu_t.start()
|
|
|
|
if not _init_done.wait(timeout=30.0):
|
|
_emit({'type': 'error', 'message': 'qemu_init timed out after 30 s'})
|
|
os._exit(1)
|
|
|
|
# Pre-register initial sensors before letting QEMU execute firmware.
|
|
for s in initial_sensors:
|
|
gpio = int(s.get('pin', 0))
|
|
sensor_type = s.get('sensor_type', '')
|
|
with _sensors_lock:
|
|
_sensors[gpio] = {
|
|
'type': sensor_type,
|
|
**{k: v for k, v in s.items() if k not in ('sensor_type', 'pin')},
|
|
'saw_low': False,
|
|
'responding': False,
|
|
}
|
|
_log(f'Pre-registered sensor {sensor_type} on GPIO {gpio}')
|
|
_sensors_ready.set()
|
|
|
|
_emit({'type': 'system', 'event': 'booted'})
|
|
_log(f'QEMU started: machine={machine} firmware={firmware_path}')
|
|
|
|
# ── 7. LEDC polling thread (100 ms interval) ──────────────────────────────
|
|
|
|
def _ledc_poll_thread() -> None:
|
|
lib.qemu_picsimlab_get_internals.restype = ctypes.c_void_p
|
|
while not _stopped.wait(0.1):
|
|
try:
|
|
ptr = lib.qemu_picsimlab_get_internals(0)
|
|
if ptr is None:
|
|
continue
|
|
arr = (ctypes.c_uint32 * 16).from_address(ptr)
|
|
for ch in range(16):
|
|
duty = int(arr[ch])
|
|
if duty > 0:
|
|
gpio = _ledc_gpio_map.get(ch, -1)
|
|
_emit({'type': 'ledc_update', 'channel': ch,
|
|
'duty': duty,
|
|
'duty_pct': round(duty / 8192 * 100, 1),
|
|
'gpio': gpio})
|
|
except Exception:
|
|
pass
|
|
|
|
threading.Thread(target=_ledc_poll_thread, daemon=True, name='ledc-poll').start()
|
|
|
|
# ── 8. Command loop (main thread reads original stdin pipe) ───────────────
|
|
|
|
for raw_line in os.fdopen(_orig_stdin_fd, 'r'):
|
|
raw_line = raw_line.strip()
|
|
if not raw_line:
|
|
continue
|
|
try:
|
|
cmd = json.loads(raw_line)
|
|
except Exception:
|
|
continue
|
|
|
|
c = cmd.get('cmd', '')
|
|
|
|
if c == 'set_pin':
|
|
# Identity pinmap: slot = gpio_num + 1
|
|
lib.qemu_picsimlab_set_pin(int(cmd['pin']) + 1, int(cmd['value']))
|
|
|
|
elif c == 'set_adc':
|
|
raw_v = int(int(cmd['millivolts']) * 4095 / 3300)
|
|
lib.qemu_picsimlab_set_apin(int(cmd['channel']), max(0, min(4095, raw_v)))
|
|
|
|
elif c == 'set_adc_raw':
|
|
lib.qemu_picsimlab_set_apin(int(cmd['channel']),
|
|
max(0, min(4095, int(cmd['raw']))))
|
|
|
|
elif c == 'uart_send':
|
|
data = base64.b64decode(cmd['data'])
|
|
buf = (ctypes.c_uint8 * len(data))(*data)
|
|
lib.qemu_picsimlab_uart_receive(int(cmd.get('uart', 0)), buf, len(data))
|
|
|
|
elif c == 'set_i2c_response':
|
|
_i2c_responses[int(cmd['addr'])] = int(cmd['response']) & 0xFF
|
|
|
|
elif c == 'set_spi_response':
|
|
_spi_response[0] = int(cmd['response']) & 0xFF
|
|
|
|
elif c == 'sensor_attach':
|
|
gpio = int(cmd['pin'])
|
|
sensor_type = cmd.get('sensor_type', '')
|
|
with _sensors_lock:
|
|
_sensors[gpio] = {
|
|
'type': sensor_type,
|
|
**{k: v for k, v in cmd.items()
|
|
if k not in ('cmd', 'pin', 'sensor_type')},
|
|
'saw_low': False,
|
|
'responding': False,
|
|
}
|
|
_log(f'Sensor {sensor_type} attached on GPIO {gpio}')
|
|
|
|
elif c == 'sensor_update':
|
|
gpio = int(cmd['pin'])
|
|
with _sensors_lock:
|
|
sensor = _sensors.get(gpio)
|
|
if sensor:
|
|
for k, v in cmd.items():
|
|
if k not in ('cmd', 'pin'):
|
|
sensor[k] = v
|
|
|
|
elif c == 'sensor_detach':
|
|
gpio = int(cmd['pin'])
|
|
with _sensors_lock:
|
|
_sensors.pop(gpio, None)
|
|
_log(f'Sensor detached from GPIO {gpio}')
|
|
|
|
elif c == 'stop':
|
|
_stopped.set()
|
|
# Signal QEMU to shut down. The assertion that fires on Windows
|
|
# ("Bail out!") is non-fatal — glib just logs it and continues.
|
|
try:
|
|
lib.qemu_cleanup()
|
|
except Exception:
|
|
pass
|
|
qemu_t.join(timeout=5.0)
|
|
# Clean up temp firmware file
|
|
if firmware_path:
|
|
try:
|
|
os.unlink(firmware_path)
|
|
except OSError:
|
|
pass
|
|
os._exit(0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|