commit eab3431993e6052691c54d223944ecde58e3e693 Author: Alessandro Mauri Date: Sun Aug 2 23:39:04 2026 +0200 initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..505a3b1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +# Python-generated files +__pycache__/ +*.py[oc] +build/ +dist/ +wheels/ +*.egg-info + +# Virtual environments +.venv diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..6324d40 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.14 diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/main.py b/main.py new file mode 100644 index 0000000..6a81c1a --- /dev/null +++ b/main.py @@ -0,0 +1,447 @@ +from asyncio.tasks import sleep +import time +from dataclasses import dataclass +from enum import Enum +from enum import IntEnum +from typing import NamedTuple +from LoRaRF import SX126x + + +LoRa = SX126x() + +# As per EN 300 220-2 V3.3.1; Annex B table 1 +# https://www.etsi.org/deliver/etsi_en/300200_300299/30022002/03.03.01_60/en_30022002v030301p.pdf +@dataclass +class BandConfig: + freq_start_mhz: float + ch_bw_khz: int + num_channels: int + max_power_mw: int + max_duty: float + polite_access: bool + + +EU868_BANDS = { + "BAND_K": BandConfig(863.0, 500, 4, 25, 0.1, True), + "BAND_L": BandConfig(865.0, 500, 6, 25, 1.0, True), + "BAND_M": BandConfig(868.0, 500, 1, 25, 1.0, True), + "BAND_N": BandConfig(868.7, 500, 1, 25, 0.1, True), + "BAND_O": BandConfig(869.4, 250, 1, 500, 10.0, True), +} + +import struct +from dataclasses import dataclass + +# Assuming Little Endian (standard for Raspberry Pi / ARM) +ENDIAN = '<' + +@dataclass +class LoRaPacketHeader: + type: int + id: int + number: int + tx_time: int # 48-bit timestamp (ms since epoch) + + def pack(self) -> bytes: + # Pack 3-byte header base + base = struct.pack(f'{ENDIAN}B B B', self.type, self.id, self.number) + # Pack 6-byte timestamp + time_bytes = self.tx_time.to_bytes(6, byteorder='little', signed=False) + return base + time_bytes + + @classmethod + def unpack(cls, data: bytes) -> LoRaPacketHeader: + t, i, n = struct.unpack(f'{ENDIAN}B B B', data[0:3]) + time_val = int.from_bytes(data[3:9], byteorder='little', signed=False) + return cls(t, i, n, time_val) + + +@dataclass +class LoRaSyncPacket: + header: LoRaPacketHeader + connected: int # 1 bit + sync_window: int # 10 bits + gs_window: int # 9 bits + security_window: int # 4 bits + + def pack(self) -> bytes: + hdr = self.header.pack() + # GCC bitfields pack LSB to MSB in Little Endian. + # Total bits = 1 + 10 + 9 + 4 = 24 bits (3 bytes). + bitfield = ( + (self.connected & 0x1) | + ((self.sync_window & 0x3FF) << 1) | + ((self.gs_window & 0x1FF) << 11) | + ((self.security_window & 0xF) << 20) + ) + bf_bytes = bitfield.to_bytes(3, byteorder='little', signed=False) + padding = b'\x00\x00\x00' + return hdr + bf_bytes + padding + + @classmethod + def unpack(cls, data: bytes) -> LoRaSyncPacket: + hdr = LoRaPacketHeader.unpack(data[:9]) + bitfield = int.from_bytes(data[9:12], byteorder='little', signed=False) + + connected = bitfield & 0x1 + sync_window = (bitfield >> 1) & 0x3FF + gs_window = (bitfield >> 11) & 0x1FF + security_window = (bitfield >> 20) & 0xF + + return cls(hdr, connected, sync_window, gs_window, security_window) + + +@dataclass +class LoRaConnectPacket: + header: LoRaPacketHeader + + def pack(self) -> bytes: + return self.header.pack() + b'\x00' * 6 + + @classmethod + def unpack(cls, data: bytes) -> LoRaConnectPacket: + return cls(LoRaPacketHeader.unpack(data[:9])) + + +@dataclass +class LoRaAcceptPacket: + header: LoRaPacketHeader + connect_time: int # 48-bit timestamp + + def pack(self) -> bytes: + hdr = self.header.pack() + ctime = self.connect_time.to_bytes(6, byteorder='little', signed=False) + return hdr + ctime + + @classmethod + def unpack(cls, data: bytes) -> LoRaAcceptPacket: + hdr = LoRaPacketHeader.unpack(data[:9]) + ctime = int.from_bytes(data[9:15], byteorder='little', signed=False) + return cls(hdr, ctime) + + +@dataclass +class LoRaDataPacket: + header: LoRaPacketHeader + # imu + imu_altitude: int # uint16_t (representing float16) + imu_vspeed: int # uint16_t (representing float16) + imu_attitude: int # uint16_t (representing float16) + imu_dt: int # int16_t + # baro + baro_p1: int # uint16_t (representing float16) + baro_p2: int # uint16_t (representing float16) + baro_dt: int # int16_t + # gps + gps_lat: float # float (32-bit IEEE 754) + gps_lon: float # float (32-bit IEEE 754) + gps_dt: int # int16_t + + def pack(self) -> bytes: + hdr = self.header.pack() + # H = uint16_t, h = int16_t, f = float (32-bit) + fmt = f'{ENDIAN}H H H h H H h f f h' + payload = struct.pack(fmt, + self.imu_altitude, self.imu_vspeed, self.imu_attitude, self.imu_dt, + self.baro_p1, self.baro_p2, self.baro_dt, + self.gps_lat, self.gps_lon, self.gps_dt) + return hdr + payload + + @classmethod + def unpack(cls, data: bytes) -> LoRaDataPacket: + hdr = LoRaPacketHeader.unpack(data[:9]) + fmt = f'{ENDIAN}H H H h H H h f f h' + payload = struct.unpack(fmt, data[9:33]) + return cls(hdr, *payload) + + +@dataclass +class LoRaCommandPacket: + header: LoRaPacketHeader + command: int + data_val: int # uint64_t + + def pack(self) -> bytes: + hdr = self.header.pack() + # B = uint8_t, Q = uint64_t + payload = struct.pack(f'{ENDIAN}B Q', self.command, self.data_val) + return hdr + payload + + @classmethod + def unpack(cls, data: bytes) -> LoRaCommandPacket: + hdr = LoRaPacketHeader.unpack(data[:9]) + cmd, val = struct.unpack(f'{ENDIAN}B Q', data[9:18]) + return cls(hdr, cmd, val) + + +# Ebyte E22-900M33S module power output, since it has an LNA and a PA, +# the actual output power is not the same as the value passed to RadioLib's +# setOutputPower() function. The table below maps the two values. +# Note that this is the output power of the module, not the effective radiated +# power (ERP) of the antenna. +# Refer to image in page 10, chapter 4.2 of the manual: https://www.cdebyte.com/products/E22-900M33S/4#Downloads +# Useful to extract data: https://plotdigitizer.com/app +# NOTE: THE X SCALE ON THIS GRAPH IS NOT CONSTANT, WHY???? +class PowerEntry(NamedTuple): + power_mw: int # The actual power in mW, for reference + radiolib_value: ( + int # The value to pass to RadioLib/LoRa module setOutputPower() + ) + +LORA_OUTPUT_POWER_TABLE = ( + PowerEntry(power_mw=59, radiolib_value=-9), + PowerEntry(power_mw=75, radiolib_value=-8), + PowerEntry(power_mw=98, radiolib_value=-7), + PowerEntry(power_mw=126, radiolib_value=-6), + PowerEntry(power_mw=173, radiolib_value=-5), + PowerEntry(power_mw=237, radiolib_value=-4), + PowerEntry(power_mw=299, radiolib_value=-3), + PowerEntry(power_mw=390, radiolib_value=-2), + PowerEntry(power_mw=487, radiolib_value=-1), + PowerEntry(power_mw=619, radiolib_value=0), + PowerEntry(power_mw=750, radiolib_value=1), + PowerEntry(power_mw=1044, radiolib_value=2), + PowerEntry(power_mw=1202, radiolib_value=3), + PowerEntry(power_mw=1409, radiolib_value=4), + PowerEntry(power_mw=1492, radiolib_value=5), + PowerEntry(power_mw=1585, radiolib_value=6), + PowerEntry(power_mw=1745, radiolib_value=7), + PowerEntry(power_mw=2218, radiolib_value=8), +) + + +"""Finds the radiolib_value for the closest supported power level in mW.""" +def get_radiolib_power(target_mw: int) -> int: + best_match = min( + LORA_OUTPUT_POWER_TABLE, key=lambda entry: abs(entry.power_mw - target_mw) + ) + return best_match.radiolib_value + + +class LoRaFCState(IntEnum): + STATE_DISCONNECTED = 0 + STATE_CONNECTING = 1 + STATE_TRANSMIT = 2 + STATE_RECEIVE = 3 + +class PacketType(IntEnum): + PKT_UNKNOWN = 0x0 + PKT_SYNC = 0x01 # Broadcast sync packet GS -> FC + PKT_CONNECT = 0x02 # Connect request packet FC -> GS + PKT_ACCEPT = 0x03 # Connect accept packet GS -> FC + PKT_DATA = 0x04 # Data packet FC -> GS + PKT_COMMAND = 0x05 # Command packet GS -> FC + + +TX_FORCE = True +LORA_FC_ID = 0xFC +LORA_GS_ID = 0xDE + +# --- State Machine Class --- +class LoRaGSStateMachine: + def __init__(self, my_id: int, fc_id: int): + # Configuration constants + self.sync_window = 1000 # ms + self.gs_window = 200 # ms + self.security_window = 10 # ms + self.MAX_SILENT_FRAMES = 5 # Define your max silent frames here + self.MY_ID = my_id + self.LORA_FC_ID = fc_id + + # Static variables from C++ + self.state = LoRaFCState.STATE_DISCONNECTED + self.sync_sent_time = 0 + self.connect_rx_time = 0 + self.packets_received = 0 + self.silent_frames = 0 + + # Global mock variables from C++ (updated by TX/RX functions) + self.last_tx_time = 0 + self.last_rx_time = 0 + + # Receive buffer + self.receive_buffer = bytearray(256) + self.receive_len = 0 + + # --- Time Helpers --- + def now_ms(self) -> int: + """Returns monotonic time in milliseconds""" + return time.monotonic_ns() // 1_000_000 + + def slot_relative_time(self, start_time_ms: int) -> int: + return self.now_ms() - start_time_ms + + + # --- The State Machine --- + def step(self) -> LoRaFCState: + """Run one iteration of the state machine. Equivalent to lora_gs_state_machine().""" + + if self.state == LoRaFCState.STATE_DISCONNECTED: + self.connect_rx_time = 0 + self.packets_received = 0 + self.silent_frames = 0 + + if (self.now_ms() - self.sync_sent_time) >= self.sync_window: + # Send sync packet to FC + hdr = LoRaPacketHeader(type=PacketType.PKT_SYNC, id=self.MY_ID, number=0, tx_time=0) + s = LoRaSyncPacket(hdr, connected=0, sync_window=self.sync_window, + gs_window=self.gs_window, security_window=self.security_window) + + if self.lora_transmit_timeout(s.pack(), self.sync_window, TX_FORCE): + self.sync_sent_time = self.last_tx_time + else: + timeout = self.sync_window - self.slot_relative_time(self.sync_sent_time) + if not self.lora_receive_timeout(timeout): + # No connect received within the sync window, return to disconnected state + return self.state + + header = self.lora_get_header() + if header.type == PacketType.PKT_CONNECT and header.id == self.LORA_FC_ID: + # Received first sync packet from GS, start the handshake + self.connect_rx_time = self.last_rx_time + self.state = LoRaFCState.STATE_CONNECTING + + elif self.state == LoRaFCState.STATE_CONNECTING: + # Send a ACCEPT packet to the GS + hdr = LoRaPacketHeader(type=PacketType.PKT_ACCEPT, id=self.MY_ID, number=0, tx_time=0) + a = LoRaAcceptPacket(hdr, connect_time=self.connect_rx_time) + + if not self.lora_transmit_timeout(a.pack(), self.sync_window // 2, TX_FORCE): + self.state = LoRaFCState.STATE_DISCONNECTED + else: + self.state = LoRaFCState.STATE_TRANSMIT + + elif self.state == LoRaFCState.STATE_TRANSMIT: + if self.silent_frames >= self.MAX_SILENT_FRAMES: + self.state = LoRaFCState.STATE_DISCONNECTED + return self.state + + # Sync window expired, need to resend sync packet + if self.slot_relative_time(self.sync_sent_time) >= self.sync_window: + hdr = LoRaPacketHeader(type=PacketType.PKT_SYNC, id=self.MY_ID, number=0, tx_time=0) + s = LoRaSyncPacket(hdr, connected=1, sync_window=self.sync_window, + gs_window=self.gs_window, security_window=self.security_window) + + if not self.lora_transmit_timeout(s.pack(), self.sync_window, TX_FORCE): + self.state = LoRaFCState.STATE_DISCONNECTED + else: + self.sync_sent_time = self.last_tx_time + return self.state + + remaining_time = self.gs_window - self.security_window - self.slot_relative_time(self.sync_sent_time) + + if remaining_time <= 0: + self.packets_received = 0 + self.state = LoRaFCState.STATE_RECEIVE + return self.state + + # Switched too early to transmit, wait for remaining listen time to expire + if remaining_time >= self.gs_window: + time.sleep((remaining_time - self.gs_window) / 1000.0) + return self.state + + # TODO: transmit commands to FC + pass + + elif self.state == LoRaFCState.STATE_RECEIVE: + remaining_time = self.sync_window - self.security_window - self.slot_relative_time(self.sync_sent_time) + + # In the transmit window + if remaining_time <= 0: + if self.packets_received == 0: + self.silent_frames += 1 + self.state = LoRaFCState.STATE_TRANSMIT + return self.state + + # In the current frame's transmit window, we shouldn't be here yet + if remaining_time >= (self.sync_window - self.gs_window + self.security_window): + if self.packets_received == 0: + self.silent_frames += 1 + self.state = LoRaFCState.STATE_TRANSMIT + return self.state + + if not self.lora_receive_timeout(remaining_time): + if self.packets_received == 0: + self.silent_frames += 1 + self.state = LoRaFCState.STATE_TRANSMIT + else: + self.packets_received += 1 + self.silent_frames = 0 + # TODO: do something with the packet + print(f"Received packet: {self.lora_get_header()}") + pass + + else: + self.state = LoRaFCState.STATE_DISCONNECTED + + return self.state + + # --- Hardware Interface Stubs --- + # You will need to implement these using LoRaRF or your chosen SPI library + + def lora_transmit_timeout(self, data: bytes, timeout_ms: int, force: bool) -> bool: + """Transmit packet over SPI to SX1262 and update self.last_tx_time.""" + # TODO: Implement via LoRaRF + LoRa.beginPacket() + LoRa.write(data, len(data)) + LoRa.endPacket(timeout_ms * 64) + self.last_tx_time = self.now_ms() + return True + + def lora_receive_timeout(self, timeout_ms: int) -> bool: + """Listen for incoming packets until timeout. Update self.last_rx_time.""" + LoRa.request(timeout_ms) + LoRa.wait() + status = LoRa.status() + + if status == LoRa.STATUS_RX_DONE: + length = LoRa.available() + if length > 0: + self.receive_buffer[:length] = LoRa.get(length) + self.receive_len = length + self.last_rx_time = self.now_ms() + return True + elif status == LoRa.STATUS_RX_TIMEOUT: + print("Reception timed out. No packet detected.") + return False + elif status == LoRa.STATUS_HEADER_ERR or status == LoRa.STATUS_CRC_ERR: + print("Packet received with errors.") + return False + return False + + def lora_get_header(self) -> LoRaPacketHeader: + """Extract and return header from the last received payload.""" + return LoRaPacketHeader.unpack(bytes(self.receive_buffer[:self.receive_len])) + +def main(): + print("Starting up LoRa module") + LoRa.setSpi(0, 0, 4_000_000) + LoRa.setPins(23, 25, 24, -1, 5) + LoRa.begin() + print("LoRa module started") + + print("Setting LoRa paramters") + band = "BAND_O" + spreading_factor = 7 + coding_rate = 5 + + LoRa.setDio2RfSwitch(True) + LoRa.setModem(SX126x.LORA_MODEM) + LoRa.setFrequency(int(EU868_BANDS[band].freq_start_mhz*1_000_000 + EU868_BANDS[band].ch_bw_khz*1000/2)) + LoRa.setBandwidth(EU868_BANDS[band].ch_bw_khz*1000) + LoRa.setTxPower(10, LoRa.TX_POWER_SX1262) + LoRa.setRxGain(LoRa.RX_GAIN_BOOSTED) + LoRa.setLoRaModulation(spreading_factor,EU868_BANDS[band].ch_bw_khz*1000, coding_rate, False) + LoRa.setHeaderType(LoRa.HEADER_EXPLICIT) + print("LoRa parameters set") + + state_machine = LoRaGSStateMachine(LORA_GS_ID, LORA_FC_ID) + while True: + state = state_machine.step() + print(f"State: {state}") + time.sleep(0.001) + + LoRa.end() + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9d462d9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,9 @@ +[project] +name = "raspberry" +version = "0.1.0" +description = "Add your description here" +readme = "README.md" +requires-python = ">=3.14" +dependencies = [ + "lorarf>=1.4.0", +] diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..8d1f7f0 --- /dev/null +++ b/uv.lock @@ -0,0 +1,39 @@ +version = 1 +revision = 3 +requires-python = ">=3.14" + +[[package]] +name = "lorarf" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "rpi-gpio" }, + { name = "spidev" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/20/0c/8d68be030540edeada07a19dd653ea6874eefedea7084f770f83fb9cdd87/LoRaRF-1.4.0.tar.gz", hash = "sha256:ac28e7d5adebfc176b7d9d0cf70a0ad8627f589c3f808664d2fce2c64f0ac9b5", size = 22412, upload-time = "2022-09-25T07:42:06.3Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e6/23/1baf7097c6a60e73b366b3770e50795847acea555548fb877831a4fc29e6/LoRaRF-1.4.0-py3-none-any.whl", hash = "sha256:8c2a918dacbdd255cefca61a91af5518e9997128083f7015114139ea14bc3c96", size = 23309, upload-time = "2022-09-25T07:42:04.36Z" }, +] + +[[package]] +name = "raspberry" +version = "0.1.0" +source = { virtual = "." } +dependencies = [ + { name = "lorarf" }, +] + +[package.metadata] +requires-dist = [{ name = "lorarf", specifier = ">=1.4.0" }] + +[[package]] +name = "rpi-gpio" +version = "0.7.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c4/0f/10b524a12b3445af1c607c27b2f5ed122ef55756e29942900e5c950735f2/RPi.GPIO-0.7.1.tar.gz", hash = "sha256:cd61c4b03c37b62bba4a5acfea9862749c33c618e0295e7e90aa4713fb373b70", size = 29090, upload-time = "2022-02-06T15:15:06.022Z" } + +[[package]] +name = "spidev" +version = "3.8" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/67/87/039b6eeea781598015b538691bc174cc0bf77df9d4d2d3b8bf9245c0de8c/spidev-3.8.tar.gz", hash = "sha256:2bc02fb8c6312d519ebf1f4331067427c0921d3f77b8bcaf05189a2e8b8382c0", size = 13893, upload-time = "2025-09-15T18:56:20.672Z" }