diff --git a/circuit-python-passthrough/Makefile b/circuit-python-passthrough/Makefile index 3355a5cb..7dc3fa9c 100644 --- a/circuit-python-passthrough/Makefile +++ b/circuit-python-passthrough/Makefile @@ -61,20 +61,24 @@ download-libraries: uv ## Download CircuitPython libraries needed for passthroug CIRCUIT_PYTHON ?= ./firmware.uf2 BOARD ?= +CIRCUIT_PYTHON_SOURCE_BOARD = $(if $(filter v5e,$(BOARD)),v5d,$(BOARD)) .PHONY: circuit-python -circuit-python: ## Download Circuit Python firmware. Usage: make circuit-python BOARD={v5a|v5b|v5c|v5d} +circuit-python: ## Download CircuitPython firmware. Usage: make circuit-python BOARD={v5a|v5b|v5c|v5d|v5e} @if [ -z "$(BOARD)" ]; then \ echo "Error: BOARD is not set"; \ - echo "Usage: make circuit-python BOARD={v5a|v5b|v5c|v5d}"; \ + echo "Usage: make circuit-python BOARD={v5a|v5b|v5c|v5d|v5e}"; \ exit 1; \ fi @case "$(BOARD)" in \ - v5a|v5b|v5c|v5d) ;; \ - *) echo "Error: Invalid board '$(BOARD)'. Must be one of: v5a, v5b, v5c, v5d"; exit 1;; \ + v5a|v5b|v5c|v5d|v5e) ;; \ + *) echo "Error: Invalid board '$(BOARD)'. Must be one of: v5a, v5b, v5c, v5d, v5e"; exit 1;; \ esac - @echo "Downloading Circuit Python firmware for board $(BOARD)..." - @curl -o $(CIRCUIT_PYTHON) -fsSL https://raw.githubusercontent.com/proveskit/flight_controller_board/main/Firmware/FC_FIRM_$(BOARD)_V1.uf2 + @if [ "$(BOARD)" = "v5e" ]; then \ + echo "Board v5e currently uses the compatible v5d CircuitPython firmware."; \ + fi + @echo "Downloading CircuitPython firmware for board $(BOARD)..." + @curl -o "$(CIRCUIT_PYTHON)" -fsSL https://raw.githubusercontent.com/proveskit/flight_controller_board/main/Firmware/FC_FIRM_$(CIRCUIT_PYTHON_SOURCE_BOARD)_V1.uf2 @echo "Firmware downloaded successfully to $(CIRCUIT_PYTHON)" .PHONY: clean diff --git a/circuit-python-passthrough/README.md b/circuit-python-passthrough/README.md index 1bea74aa..e6fbf752 100644 --- a/circuit-python-passthrough/README.md +++ b/circuit-python-passthrough/README.md @@ -2,8 +2,22 @@ This folder holds the code necessary to set up a flight controller board to act as a GDS ground station passthrough. To install to a board, cd into this directory and run: -``` +```sh make install {sband|lora} BOARD_MOUNT_POINT={path to board} ``` -You can also use the `make circuit-python` command to get the CircuitPython firmware for the board you are using. +Power-cycle the board after the first install. The USB data serial endpoint is +enabled by `boot.py` and is not created by a CircuitPython soft reload. + +The LoRa passthrough auto-detects the SX1276 used on boards through v5d and +the SX1262-based E22-400M30S used on v5e. + +Download the CircuitPython firmware for a board with: + +```sh +make circuit-python BOARD={v5a|v5b|v5c|v5d|v5e} +``` + +The v5e command currently downloads the compatible v5d CircuitPython image. +The LoRa passthrough accesses the v5e radio GPIOs directly, so it does not +depend on v5e-specific CircuitPython board aliases. diff --git a/circuit-python-passthrough/code-lora.py b/circuit-python-passthrough/code-lora.py index 4460d807..5019ff2f 100644 --- a/circuit-python-passthrough/code-lora.py +++ b/circuit-python-passthrough/code-lora.py @@ -1,8 +1,7 @@ -""" -CircuitPython Feather RP2350 LoRa Radio forwarder +"""CircuitPython LoRa radio passthrough for PROVES flight controller boards. -This code will forward any received LoRa packets to the serial console (sys.stdout). It cycles through neo pixel colors -to indicate packet reception. +The passthrough supports the SX1276 used through v5d and the SX1262-based +E22-400M30S used on v5e. """ import time @@ -10,10 +9,13 @@ import adafruit_rfm9x import board import digitalio +import microcontroller # Turn off auto-reload to prevent LoRa module reset import supervisor import usb_cdc +from proves_sx126._sx126x import ERR_NONE, ERR_RX_TIMEOUT, SX126X_REG_RX_GAIN +from proves_sx126.sx1262 import SX1262 supervisor.runtime.autoreload = False @@ -25,18 +27,246 @@ "U": (7, 500000, 125000, 0), # Uplink only } +FREQUENCY_MHZ = 437.4 +CODING_RATE = 5 +PREAMBLE_LENGTH = 8 +FPRIME_LORA_HEADER = b"\x00\x00\x00\x00" +SX1262_MAX_POWER_DBM = 22 +SX1262_RX_GAIN_BOOSTED = 0x96 +SX1262_TCXO_STARTUP_DELAY_US = 10000 + + +class CircuitPythonIRQ: + """Adapt CircuitPython's DIO value property to the SX126 driver's API.""" + + def __init__(self, digital_input: digitalio.DigitalInOut): + self._digital_input = digital_input + + def value(self) -> bool: + """Return the current DIO input value.""" + return self._digital_input.value + + +class CircuitPythonSX1262(SX1262): + """SX1262 compatibility wrapper for CircuitPython DigitalInOut.""" + + def setBlockingCallback(self, blocking: bool, callback=None) -> int: + """Configure blocking mode without recreating an allocated DIO1 pin.""" + if not blocking: + return super().setBlockingCallback(blocking, callback) + + self.blocking = True + self._callbackFunction = self._dummyFunction + return self.standby() + + +class SX1276Radio: + """SX1276 backend used by PROVES flight controller boards through v5d.""" + + def __init__(self, freq_mhz: float): + chip_select = digitalio.DigitalInOut(microcontroller.pin.GPIO9) + reset = digitalio.DigitalInOut(microcontroller.pin.GPIO6) + + try: + self._radio = adafruit_rfm9x.RFM9x( + board.SPI(), chip_select, reset, freq_mhz + ) + except Exception: + chip_select.deinit() + reset.deinit() + raise + + self._radio.coding_rate = CODING_RATE + self._radio.preamble_length = PREAMBLE_LENGTH + + @property + def spreading_factor(self) -> int: + """Return the configured spreading factor.""" + return self._radio.spreading_factor + + @spreading_factor.setter + def spreading_factor(self, value: int) -> None: + self._radio.spreading_factor = value + + @property + def signal_bandwidth(self) -> int: + """Return the configured signal bandwidth in Hz.""" + return self._radio.signal_bandwidth + + @signal_bandwidth.setter + def signal_bandwidth(self, value: int) -> None: + self._radio.signal_bandwidth = value + + def send(self, data: bytes) -> bool: + """Transmit a packet.""" + return self._radio.send(data) + + def receive(self, timeout: float) -> bytes | None: + """Receive a packet.""" + return self._radio.receive(timeout=timeout) + + def idle(self) -> None: + """Put the radio in standby.""" + self._radio.idle() + + +class SX1262Radio: + """SX1262/E22-400M30S backend used by the v5e board.""" + + def __init__(self, freq_mhz: float): + # v5e device tree: SPI1 SCK/MOSI/MISO = GPIO 10/11/12. + self._spi = board.SPI() + self._chip_select = self._output_pin(microcontroller.pin.GPIO9, True) + self._reset = self._output_pin(microcontroller.pin.GPIO6, True) + self._busy = self._input_pin(microcontroller.pin.GPIO13) + self._dio1 = self._input_pin(microcontroller.pin.GPIO14) + self._tx_enable = self._output_pin(microcontroller.pin.GPIO21, False) + self._rx_enable = self._output_pin(microcontroller.pin.GPIO22, False) + + self._radio = CircuitPythonSX1262( + self._spi, + self._chip_select, + self._dio1, + self._reset, + self._busy, + ) + self._radio.irq = CircuitPythonIRQ(self._dio1) + self._spreading_factor = RADIO_PARAMS["1"][0] + self._signal_bandwidth = RADIO_PARAMS["1"][1] + + status = self._radio.begin( + freq=freq_mhz, + bw=self._signal_bandwidth / 1000, + sf=self._spreading_factor, + cr=CODING_RATE, + power=SX1262_MAX_POWER_DBM, + preambleLength=PREAMBLE_LENGTH, + crcOn=True, + implicit=False, + syncWord=0x12, + tcxoVoltage=1.8, + blocking=True, + ) + self._check_status(status, "initialize") + self._check_status( + self._radio.setTCXO(1.8, delay=SX1262_TCXO_STARTUP_DELAY_US), + "configure TCXO startup delay", + ) + self._check_status( + self._radio.setDio2AsRfSwitch(False), "disable DIO2 RF switching" + ) + self._check_status( + self._radio.writeRegister(SX126X_REG_RX_GAIN, [SX1262_RX_GAIN_BOOSTED], 1), + "enable boosted RX gain", + ) + + @staticmethod + def _input_pin(pin: microcontroller.Pin) -> digitalio.DigitalInOut: + digital_pin = digitalio.DigitalInOut(pin) + digital_pin.switch_to_input() + return digital_pin + + @staticmethod + def _output_pin(pin: microcontroller.Pin, value: bool) -> digitalio.DigitalInOut: + digital_pin = digitalio.DigitalInOut(pin) + digital_pin.switch_to_output(value=value) + return digital_pin + + def _check_status(self, status: int, operation: str) -> None: + if status != ERR_NONE: + error = self._radio.STATUS.get(status, str(status)) + raise RuntimeError(f"SX1262 failed to {operation}: {error}") + + def _set_rf_path(self, transmit: bool = False, receive: bool = False) -> None: + # Never enable both paths at once. The DTS marks both enables active-high. + self._tx_enable.value = False + self._rx_enable.value = False + self._tx_enable.value = transmit + self._rx_enable.value = receive + + @property + def spreading_factor(self) -> int: + """Return the configured spreading factor.""" + return self._spreading_factor + + @spreading_factor.setter + def spreading_factor(self, value: int) -> None: + self._check_status(self._radio.setSpreadingFactor(value), "set SF") + self._spreading_factor = value + + @property + def signal_bandwidth(self) -> int: + """Return the configured signal bandwidth in Hz.""" + return self._signal_bandwidth + + @signal_bandwidth.setter + def signal_bandwidth(self, value: int) -> None: + self._check_status(self._radio.setBandwidth(value / 1000), "set bandwidth") + self._signal_bandwidth = value + + def send(self, data: bytes) -> bool: + """Transmit a packet using the external v5e TX RF path.""" + self._set_rf_path(transmit=True) + try: + # F Prime expects the four-byte header that adafruit_rfm9x adds + # automatically for the legacy SX1276 passthrough. + _, status = self._radio.send(FPRIME_LORA_HEADER + data) + if status != ERR_NONE: + error = self._radio.STATUS.get(status, str(status)) + print(f"[ERROR] SX1262 transmit failed: {error}") + return False + return True + finally: + self._set_rf_path() + + def receive(self, timeout: float) -> bytes | None: + """Receive a packet using the external v5e RX RF path.""" + self._set_rf_path(receive=True) + try: + received, status = self._radio.recv( + timeout_en=True, timeout_ms=int(timeout * 1000) + ) + finally: + self._set_rf_path() + + if status == ERR_NONE: + if len(received) < len(FPRIME_LORA_HEADER): + print("[ERROR] SX1262 received a packet shorter than the LoRa header") + return None + # Match adafruit_rfm9x.receive(), which strips its RadioHead + # header before returning bytes to the USB/GDS transport. + return received[len(FPRIME_LORA_HEADER) :] + if status != ERR_RX_TIMEOUT: + error = self._radio.STATUS.get(status, str(status)) + print(f"[ERROR] SX1262 receive failed: {error}") + return None + + def idle(self) -> None: + """Disable both external RF paths and put the radio in standby.""" + self._set_rf_path() + self._check_status(self._radio.standby(), "enter standby") + class Lora(object): """LoRa Radio class""" - CS = digitalio.DigitalInOut(board.SPI0_CS0) - RESET = digitalio.DigitalInOut(board.RF1_RST) - - def __init__(self, freq_mhz: float = 437.4): + def __init__(self, freq_mhz: float = FREQUENCY_MHZ): """Initialize the LoRa module""" - self.rfm95 = adafruit_rfm9x.RFM9x(board.SPI(), self.CS, self.RESET, freq_mhz) - self.rfm95.coding_rate = 5 - self.rfm95.preamble_length = 8 + board_id = getattr(board, "board_id", "").lower() + if board_id.endswith("v5e"): + self.radio = SX1262Radio(freq_mhz) + radio_name = "SX1262" + else: + try: + self.radio = SX1276Radio(freq_mhz) + radio_name = "SX1276" + except RuntimeError: + # A v5d CircuitPython build can be used on v5e because the + # passthrough references the v5e GPIOs directly. + self.radio = SX1262Radio(freq_mhz) + radio_name = "SX1262" + + print(f"[INFO] Initialized {radio_name} LoRa radio") self.mode = "1" self.up_count = 0 self.dw_count = 0 @@ -46,20 +276,21 @@ def transmit(self, data: bytes) -> None: if data is None or len(data) == 0: return _, up_bandwidth, down_bandwidth, _ = RADIO_PARAMS[self._mode] - self.rfm95.signal_bandwidth = up_bandwidth - success = self.rfm95.send(data) + self.radio.signal_bandwidth = up_bandwidth + success = self.radio.send(data) if not success: print("[ERROR] Failed to transmit packet") - self.up_count += 1 - self.rfm95.signal_bandwidth = down_bandwidth + else: + self.up_count += 1 + self.radio.signal_bandwidth = down_bandwidth def receive(self) -> bytes: """Receive data over LoRa""" if self._mode == "U": return b"" _, _, dw_bandwidth, timeout = RADIO_PARAMS[self._mode] - self.rfm95.signal_bandwidth = dw_bandwidth - received = self.rfm95.receive(timeout=timeout) + self.radio.signal_bandwidth = dw_bandwidth + received = self.radio.receive(timeout=timeout) if received is not None: self.dw_count += 1 return received @@ -70,7 +301,7 @@ def dump(self): _, up_bandwidth, dw_bandwidth, _ = RADIO_PARAMS[self._mode] print(f"[INFO] Uplink: {self.up_count}, Downlink: {self.dw_count}") print( - f"[INFO] Spreading factor: {self.rfm95.spreading_factor}, bandwidth up: {up_bandwidth}, bandwidth down: {dw_bandwidth}" + f"[INFO] Spreading factor: {self.radio.spreading_factor}, bandwidth up: {up_bandwidth}, bandwidth down: {dw_bandwidth}" ) @property @@ -85,15 +316,21 @@ def mode(self, value: str) -> None: raise ValueError(f"{value} is not valid Radio setting") self._mode = value spreading, _, _, _ = RADIO_PARAMS[value] - self.rfm95.spreading_factor = spreading + self.radio.spreading_factor = spreading + if value == "U": + self.radio.idle() -usb_cdc.console.timeout = 0.01 -usb_cdc.data.timeout = 0.01 data_console = b"" print("[INFO] LoRa Receiver receiving packets") lora = Lora() +data_serial = usb_cdc.data +if data_serial is None: + print( + "[WARNING] USB data serial is unavailable. Power-cycle the board " + "to apply boot.py; a soft reload is not sufficient." + ) last_time = time.time() while True: @@ -102,15 +339,21 @@ def mode(self, value: str) -> None: lora.dump() # Read data from USB CDC and transmit over LoRa - try: - packet_data = usb_cdc.data.read(usb_cdc.data.in_waiting) - if packet_data is not None and len(packet_data) > 0: - lora.transmit(packet_data) - continue # Most important to prioritize transmitting over receiving - except Exception as e: - print(f"[ERROR] {e} transmitting packet of size {len(packet_data)}") + packet_data = b"" + if data_serial is not None: + try: + if data_serial.in_waiting > 0: + packet_data = data_serial.read(data_serial.in_waiting) + if packet_data: + lora.transmit(packet_data) + continue # Most important to prioritize transmitting over receiving + except Exception as e: + print(f"[ERROR] {e} transmitting packet of size {len(packet_data)}") + # Read data from LoRa and write to USB CDC - packet = usb_cdc.data.write(lora.receive()) + packet = lora.receive() + if packet and data_serial is not None: + data_serial.write(packet) # Read data from console to change radio parameters if usb_cdc.console.in_waiting > 0: diff --git a/circuit-python-passthrough/requirements.txt b/circuit-python-passthrough/requirements.txt index 1b9f5927..4442777d 100644 --- a/circuit-python-passthrough/requirements.txt +++ b/circuit-python-passthrough/requirements.txt @@ -1,4 +1,5 @@ adafruit-circuitpython-rfm9x==2.2.22 adafruit-circuitpython-rfm==1.0.3 Adafruit_CircuitPython_MCP230xx==2.5.19 +proves-circuitpython-sx126 @ git+https://github.com/proveskit/micropySX126X@1.0.0 proves-circuitpython-sx1280 @ git+https://github.com/proveskit/CircuitPython_SX1280@1.0.3