Improved otaa and cleanup code a bit

jeroennijhof · Jun 4, 2017 · 6aa27ca · 6aa27ca
1 parent 5cf4eea
commit 6aa27ca
Show file tree

Hide file tree

Showing 5 changed files with 52 additions and 124 deletions.
diff --git a/LoRaWAN/JoinAcceptPayload.py b/LoRaWAN/JoinAcceptPayload.py
@@ -31,7 +31,7 @@ def get_netid(self):
         return self.netid
 
     def get_devaddr(self):
-        return self.devaddr
+        return list(map(int, reversed(self.devaddr)))
 
     def get_dlsettings(self):
         return self.dlsettings
@@ -78,22 +78,22 @@ def encrypt_payload(self, key, direction, mhdr):
         cipher = AES.new(bytes(key))
         return list(map(int, cipher.decrypt(bytes(a))))
 
-    def derive_nwkey(self, key, devnonce):
+    def derive_nwskey(self, key, devnonce):
         a = [0x01]
         a += self.get_appnonce()
         a += self.get_netid()
         a += devnonce
         a += [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
 
         cipher = AES.new(bytes(key))
-        return list(map(hex, cipher.encrypt(bytes(a))))
+        return list(map(int, cipher.encrypt(bytes(a))))
 
-    def derive_appkey(self, key, devnonce):
+    def derive_appskey(self, key, devnonce):
         a = [0x02]
         a += self.get_appnonce()
         a += self.get_netid()
         a += devnonce
         a += [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
 
         cipher = AES.new(bytes(key))
-        return list(map(hex, cipher.encrypt(bytes(a))))
+        return list(map(int, cipher.encrypt(bytes(a))))
diff --git a/LoRaWAN/PhyPayload.py b/LoRaWAN/PhyPayload.py
@@ -76,11 +76,17 @@ def valid_mic(self):
         else:
             return self.get_mic() == self.mac_payload.frm_payload.compute_mic(self.nwkey, self.get_direction(), self.get_mhdr())
 
+    def get_devaddr(self):
+        if self.get_mhdr().get_mtype() == MHDR.JOIN_ACCEPT:
+            return self.mac_payload.frm_payload.get_devaddr()
+        else:
+            return self.mac_payload.fhdr.get_devaddr()
+
     def get_payload(self):
         return self.mac_payload.frm_payload.decrypt_payload(self.appkey, self.get_direction(), self.mic)
 
-    def derive_nwkey(self, devnonce):
-        return self.mac_payload.frm_payload.derive_nwkey(self.appkey, devnonce)
+    def derive_nwskey(self, devnonce):
+        return self.mac_payload.frm_payload.derive_nwskey(self.appkey, devnonce)
 
-    def derive_appkey(self, devnonce):
-        return self.mac_payload.frm_payload.derive_appkey(self.appkey, devnonce)
+    def derive_appskey(self, devnonce):
+        return self.mac_payload.frm_payload.derive_appskey(self.appkey, devnonce)
diff --git a/otaa_ttn.py b/otaa_ttn.py
@@ -6,17 +6,14 @@
 from SX127x.board_config import BOARD
 import LoRaWAN
 from LoRaWAN.MHDR import MHDR
+from random import randrange
 
 BOARD.setup()
 parser = LoRaArgumentParser("LoRaWAN sender")
 
-class LoRaWANsend(LoRa):
-    def __init__(self, deveui = [], appeui = [], appkey = [], devnonce = [], verbose = False):
-        super(LoRaWANsend, self).__init__(verbose)
-        self.deveui = deveui
-        self.appeui = appeui
-        self.appkey = appkey
-        self.devnonce = devnonce
+class LoRaWANotaa(LoRa):
+    def __init__(self, verbose = False):
+        super(LoRaWANotaa, self).__init__(verbose)
 
     def on_rx_done(self):
         print("RxDone")
@@ -28,14 +25,17 @@ def on_rx_done(self):
         lorawan.read(payload)
         print(lorawan.get_payload())
         print(lorawan.get_mhdr().get_mversion())
-        print(lorawan.get_mhdr().get_mtype())
-        print(lorawan.get_mic())
-        print(lorawan.compute_mic())
-        print(lorawan.valid_mic())
-        print(lorawan.derive_nwkey(devnonce))
-        print(lorawan.derive_appkey(devnonce))
-        print("\n")
-        sys.exit(0)
+
+        if lorawan.get_mhdr().get_mtype() == MHDR.JOIN_ACCEPT:
+            print("Got LoRaWAN join accept")
+            print(lorawan.valid_mic())
+            print(lorawan.get_devaddr())
+            print(lorawan.derive_nwskey(devnonce))
+            print(lorawan.derive_appskey(devnonce))
+            print("\n")
+            sys.exit(0)
+
+        print("Got LoRaWAN message continue listen for join accept")
 
     def on_tx_done(self):
         self.clear_irq_flags(TxDone=1)
@@ -47,31 +47,11 @@ def on_tx_done(self):
         self.reset_ptr_rx()
         self.set_mode(MODE.RXCONT)
 
-    def on_cad_done(self):
-        print("on_CadDone")
-        print(self.get_irq_flags())
-
-    def on_rx_timeout(self):
-        print("on_RxTimeout")
-        print(self.get_irq_flags())
-
-    def on_valid_header(self):
-        print("on_ValidHeader")
-        print(self.get_irq_flags())
-
-    def on_payload_crc_error(self):
-        print("on_PayloadCrcError")
-        print(self.get_irq_flags())
-
-    def on_fhss_change_channel(self):
-        print("on_FhssChangeChannel")
-        print(self.get_irq_flags())
-
     def start(self):
         self.tx_counter = 1
 
-        lorawan = LoRaWAN.new(self.appkey)
-        lorawan.create(MHDR.JOIN_REQUEST, {'deveui': self.deveui, 'appeui': self.appeui, 'devnonce': self.devnonce})
+        lorawan = LoRaWAN.new(appkey)
+        lorawan.create(MHDR.JOIN_REQUEST, {'deveui': deveui, 'appeui': appeui, 'devnonce': devnonce})
 
         self.write_payload(lorawan.to_raw())
         self.set_mode(MODE.TX)
@@ -80,11 +60,11 @@ def start(self):
 
 
 # Init
-deveui = [0x00, 0x82, 0xAA, 0x0D, 0x42, 0x9C, 0x79, 0x34]
-appeui = [0x70, 0xB3, 0xD5, 0x7E, 0xF0, 0x00, 0x4D, 0xBC]
-appkey = [0x13, 0x1C, 0x8A, 0xF7, 0xA3, 0xE4, 0x35, 0xD0, 0xD5, 0xE9, 0x47, 0x6B, 0x04, 0xB9, 0x16, 0x39]
-devnonce = [0x01, 0x25]
-lora = LoRaWANsend(deveui, appeui, appkey, devnonce, False)
+deveui = [0x00, 0x47, 0x64, 0xB1, 0xAB, 0xC6, 0x4F, 0x7C]
+appeui = [0x70, 0xB3, 0xD5, 0x7E, 0xF0, 0x00, 0x51, 0x34]
+appkey = [0xA1, 0x0F, 0x0E, 0x87, 0x0A, 0x15, 0x58, 0x40, 0x89, 0x73, 0xC0, 0x60, 0x1E, 0x19, 0xC3, 0xD1]
+devnonce = [randrange(256), randrange(256)]
+lora = LoRaWANotaa(False)
 
 # Setup
 lora.set_mode(MODE.SLEEP)
@@ -100,14 +80,12 @@ def start(self):
 assert(lora.get_agc_auto_on() == 1)
 
 try:
-    print("Sending LoRaWAN message\n")
+    print("Sending LoRaWAN join request\n")
     lora.start()
 except KeyboardInterrupt:
     sys.stdout.flush()
     print("\nKeyboardInterrupt")
 finally:
     sys.stdout.flush()
-    print("")
     lora.set_mode(MODE.SLEEP)
-    print(lora)
     BOARD.teardown()
diff --git a/rx_ttn.py b/rx_ttn.py
@@ -9,12 +9,8 @@
 parser = LoRaArgumentParser("LoRaWAN receiver")
 
 class LoRaWANrcv(LoRa):
-    def __init__(self, nwkey = [], appkey = [], verbose = False):
+    def __init__(self, verbose = False):
         super(LoRaWANrcv, self).__init__(verbose)
-        self.set_mode(MODE.SLEEP)
-        self.set_dio_mapping([0] * 6)
-        self.nwkey = nwkey
-        self.appkey = appkey
 
     def on_rx_done(self):
         print("RxDone")
@@ -23,7 +19,7 @@ def on_rx_done(self):
         payload = self.read_payload(nocheck=True)
         print("".join(format(x, '02x') for x in bytes(payload)))
 
-        lorawan = LoRaWAN.new(self.nwkey, self.appkey)
+        lorawan = LoRaWAN.new(nwskey, appskey)
         lorawan.read(payload)
         print(lorawan.get_mhdr().get_mversion())
         print(lorawan.get_mhdr().get_mtype())
@@ -37,30 +33,6 @@ def on_rx_done(self):
         self.reset_ptr_rx()
         self.set_mode(MODE.RXCONT)
 
-    def on_tx_done(self):
-        print("TxDone")
-        print(self.get_irq_flags())
-
-    def on_cad_done(self):
-        print("on_CadDone")
-        print(self.get_irq_flags())
-
-    def on_rx_timeout(self):
-        print("on_RxTimeout")
-        print(self.get_irq_flags())
-
-    def on_valid_header(self):
-        print("on_ValidHeader")
-        print(self.get_irq_flags())
-
-    def on_payload_crc_error(self):
-        print("on_PayloadCrcError")
-        print(self.get_irq_flags())
-
-    def on_fhss_change_channel(self):
-        print("on_FhssChangeChannel")
-        print(self.get_irq_flags())
-
     def start(self):
         self.reset_ptr_rx()
         self.set_mode(MODE.RXCONT)
@@ -69,11 +41,13 @@ def start(self):
 
 
 # Init
-nwkey = [0xC3, 0x24, 0x64, 0x98, 0xDE, 0x56, 0x5D, 0x8C, 0x55, 0x88, 0x7C, 0x05, 0x86, 0xF9, 0x82, 0x26]
-appkey = [0x15, 0xF6, 0xF4, 0xD4, 0x2A, 0x95, 0xB0, 0x97, 0x53, 0x27, 0xB7, 0xC1, 0x45, 0x6E, 0xC5, 0x45]
-lora = LoRaWANrcv(nwkey, appkey, False)
+nwskey = [0xC3, 0x24, 0x64, 0x98, 0xDE, 0x56, 0x5D, 0x8C, 0x55, 0x88, 0x7C, 0x05, 0x86, 0xF9, 0x82, 0x26]
+appskey = [0x15, 0xF6, 0xF4, 0xD4, 0x2A, 0x95, 0xB0, 0x97, 0x53, 0x27, 0xB7, 0xC1, 0x45, 0x6E, 0xC5, 0x45]
+lora = LoRaWANrcv(False)
 
 # Setup
+lora.set_mode(MODE.SLEEP)
+lora.set_dio_mapping([0] * 6)
 lora.set_freq(868.1)
 lora.set_pa_config(pa_select=1)
 lora.set_spreading_factor(7)
@@ -91,7 +65,5 @@ def start(self):
     print("\nKeyboardInterrupt")
 finally:
     sys.stdout.flush()
-    print("")
     lora.set_mode(MODE.SLEEP)
-    print(lora)
     BOARD.teardown()
diff --git a/tx_ttn.py b/tx_ttn.py
@@ -13,47 +13,19 @@
 class LoRaWANsend(LoRa):
     def __init__(self, devaddr = [], nwkey = [], appkey = [], verbose = False):
         super(LoRaWANsend, self).__init__(verbose)
-        self.set_mode(MODE.SLEEP)
-        self.set_dio_mapping([1,0,0,0,0,0])
         self.devaddr = devaddr
         self.nwkey = nwkey
         self.appkey = appkey
 
-    def on_rx_done(self):
-        print("RxDone")
-        print(self.get_irq_flags())
-
     def on_tx_done(self):
         self.set_mode(MODE.STDBY)
         self.clear_irq_flags(TxDone=1)
         print("TxDone")
         sys.exit(0)
 
-    def on_cad_done(self):
-        print("on_CadDone")
-        print(self.get_irq_flags())
-
-    def on_rx_timeout(self):
-        print("on_RxTimeout")
-        print(self.get_irq_flags())
-
-    def on_valid_header(self):
-        print("on_ValidHeader")
-        print(self.get_irq_flags())
-
-    def on_payload_crc_error(self):
-        print("on_PayloadCrcError")
-        print(self.get_irq_flags())
-
-    def on_fhss_change_channel(self):
-        print("on_FhssChangeChannel")
-        print(self.get_irq_flags())
-
     def start(self):
-        self.tx_counter = 1
-
-        lorawan = LoRaWAN.new(self.nwkey, self.appkey)
-        lorawan.create(MHDR.UNCONF_DATA_UP, {'devaddr': self.devaddr, 'fcnt': self.tx_counter, 'data': list(map(ord, 'Python rules!')) })
+        lorawan = LoRaWAN.new(nwskey, appskey)
+        lorawan.create(MHDR.UNCONF_DATA_UP, {'devaddr': devaddr, 'fcnt': 1, 'data': list(map(ord, 'Python rules!')) })
 
         self.write_payload(lorawan.to_raw())
         self.set_mode(MODE.TX)
@@ -63,11 +35,13 @@ def start(self):
 
 # Init
 devaddr = [0x26, 0x01, 0x11, 0x5F]
-nwkey = [0xC3, 0x24, 0x64, 0x98, 0xDE, 0x56, 0x5D, 0x8C, 0x55, 0x88, 0x7C, 0x05, 0x86, 0xF9, 0x82, 0x26]
-appkey = [0x15, 0xF6, 0xF4, 0xD4, 0x2A, 0x95, 0xB0, 0x97, 0x53, 0x27, 0xB7, 0xC1, 0x45, 0x6E, 0xC5, 0x45]
-lora = LoRaWANsend(devaddr, nwkey, appkey, False)
+nwskey = [0xC3, 0x24, 0x64, 0x98, 0xDE, 0x56, 0x5D, 0x8C, 0x55, 0x88, 0x7C, 0x05, 0x86, 0xF9, 0x82, 0x26]
+appskey = [0x15, 0xF6, 0xF4, 0xD4, 0x2A, 0x95, 0xB0, 0x97, 0x53, 0x27, 0xB7, 0xC1, 0x45, 0x6E, 0xC5, 0x45]
+lora = LoRaWANsend(False)
 
 # Setup
+lora.set_mode(MODE.SLEEP)
+lora.set_dio_mapping([1,0,0,0,0,0])
 lora.set_freq(868.1)
 lora.set_pa_config(pa_select=1)
 lora.set_spreading_factor(7)
@@ -86,7 +60,5 @@ def start(self):
     print("\nKeyboardInterrupt")
 finally:
     sys.stdout.flush()
-    print("")
     lora.set_mode(MODE.SLEEP)
-    print(lora)
     BOARD.teardown()