from crypt_random import secure_random from lora_handler import LoRaHandler, LoRaMessage import time import uasyncio as asyncio import json from machine import Pin led = Pin("LED", Pin.OUT) class Connection: def __init__(self, bluetooth_handler=None): # ensure led is off during initialization to indicate we're not ready yet led.value(0) try: with open("settings.json", "r") as f: self.settings = json.load(f) except FileNotFoundError: print("No settings file found. Using defaults...") self.settings = { "display_name": "Unnamed Node" } self.bluetooth_handler = bluetooth_handler self.load_public_constants() # generate diffie-hellman keys print("Generating keys...") print(" - Generating private key...") self.private_key = int.from_bytes(secure_random(32), "big") # 32 bytes = 256 bits print(" - Generating public key...") self.public_key = pow(self.G, self.private_key, self.P) # start lora handler self.lora = LoRaHandler() # turn on status LED to indicate we're ready led.value(1) async def bluetooth_listener(self): while True: msg_type, payload = await self.bluetooth_handler.packets.get() if msg_type == 3: # advertise self self.lora.advertise() def bluetooth_send(self, msg_type: int, payload: dict): if self.bluetooth_handler: self.bluetooth_handler.send_packet(msg_type, payload) async def start(self): print("Starting connection...") if self.bluetooth_handler: asyncio.create_task(self.bluetooth_listener()) while True: msg: LoRaMessage = await self.lora.message_queue.get() print(f"Received message from {msg.sender_id}: {msg.payload}") if msg.payload == b"Ping!": self.bluetooth_send(4, {"id": msg.sender_id, "name": self.settings["display_name"]}) def load_public_constants(self): with open("public_constants.txt", "r") as f: self.P = int(f.readline(), 16) self.G = int(f.readline(), 16)