refactor for better code readability

bleScanner/BleHelper.py

@@ -1,4 +1,5 @@
 #!/usr/bin/python3
+#perform ble advertisment scan
 import os
 import sys
 import struct
@@ -7,7 +8,7 @@
 
 class BleHelper:
     def __init__(self,dev_id = 0):
-
+        # bt parameter codes
         self.LE_META_EVENT = 0x3e
         self.LE_PUBLIC_ADDRESS = 0x00
         self.LE_RANDOM_ADDRESS = 0x01
@@ -16,104 +17,63 @@ def __init__(self,dev_id = 0):
         self.OCF_LE_SET_SCAN_PARAMETERS = 0x000B
         self.OCF_LE_SET_SCAN_ENABLE = 0x000C
         self.OCF_LE_CREATE_CONN = 0x000D
-
+        
         self.LE_ROLE_MASTER = 0x00
         self.LE_ROLE_SLAVE = 0x01
-
         # these are actually subevents of LE_META_EVENT
         self.EVT_LE_CONN_COMPLETE = 0x01
         self.EVT_LE_ADVERTISING_REPORT = 0x02
         self.EVT_LE_CONN_UPDATE_COMPLETE = 0x03
         self.EVT_LE_READ_REMOTE_USED_FEATURES_COMPLETE = 0x04
-
         # Advertisment event types
         self.ADV_IND = 0x00
         self.ADV_DIRECT_IND = 0x01
         self.ADV_SCAN_IND = 0x02
         self.ADV_NONCONN_IND = 0x03
         self.ADV_SCAN_RSP = 0x04
+        #try to start scanner
+        self._run_ble_scanner()
+        self._hci_le_set_scan_parameters(self.sock)
+        self._hci_enable_le_scan(self.sock)
+
+    def _run_ble_scanner(self):
         try:
             self.sock = bluez.hci_open_dev(dev_id)
             print("ble thread started")
 
-        except:
+        except Exception as e:
             print("error accessing bluetooth device...")
+            print(e)
             sys.exit(1)
 
-        self.hci_le_set_scan_parameters(self.sock)
-        self.hci_enable_le_scan(self.sock)
-
-
-    def returnnumberpacket(self, pkt):
-        myInteger = 0
-        multiple = 256
-        for c in pkt:
-            myInteger += struct.unpack("B", c)[0] * multiple
-            multiple = 1
-        return myInteger
-
-
-    def returnstringpacket(self,pkt):
-        myString = ""
-        for c in pkt:
-            myString += "%02x" % struct.unpack("B", c)[0]
-        return myString
-
 
-    def get_packed_bdaddr(self, bdaddr_string):
-        packable_addr = []
-        addr = bdaddr_string.split(':')
-        addr.reverse()
-        for b in addr:
-            packable_addr.append(int(b, 16))
-        return struct.pack("<BBBBBB", *packable_addr)
+    def _hci_enable_le_scan(self, sock):
+        """enable bt device ble scan"""
+        self._hci_toggle_le_scan(sock, 0x01)
 
 
-    def packed_bdaddr_to_string(self, bdaddr_packed):
-        return ':'.join('%02x' % i for i in struct.unpack("<BBBBBB", bdaddr_packed[::-1]))
+    def _hci_disable_le_scan(self, sock):
+        """disable bt device ble scan"""
+        self._hci_toggle_le_scan(sock, 0x00)
 
 
-    def hci_enable_le_scan(self, sock):
-        self.hci_toggle_le_scan(sock, 0x01)
-
-
-    def hci_disable_le_scan(self, sock):
-        self.hci_toggle_le_scan(sock, 0x00)
-
-
-    def hci_toggle_le_scan(self, sock, enable):
-        # hci_le_set_scan_enable(dd, 0x01, filter_dup, 1000);
-        # memset(&scan_cp, 0, sizeof(scan_cp));
-        # uint8_t         enable;
-        #       uint8_t         filter_dup;
-        #        scan_cp.enable = enable;
-        #        scan_cp.filter_dup = filter_dup;
-        #
-        #        memset(&rq, 0, sizeof(rq));
-        #        rq.ogf = OGF_LE_CTL;
-        #        rq.ocf = OCF_LE_SET_SCAN_ENABLE;
-        #        rq.cparam = &scan_cp;
-        #        rq.clen = LE_SET_SCAN_ENABLE_CP_SIZE;
-        #        rq.rparam = &status;
-        #        rq.rlen = 1;
-
-        #        if (hci_send_req(dd, &rq, to) < 0)
-        #                return -1;
+    def _hci_toggle_le_scan(self, sock, enable):
+        """set ble scanner state"""
         cmd_pkt = struct.pack("<BB", enable, 0x00)
         bluez.hci_send_cmd(sock, self.OGF_LE_CTL, self.OCF_LE_SET_SCAN_ENABLE, cmd_pkt)
 
 
-    def hci_le_set_scan_parameters(self, sock):
+    def _hci_le_set_scan_parameters(self, sock):
+        """set ble scanner parameters"""
         old_filter = sock.getsockopt(bluez.SOL_HCI, bluez.HCI_FILTER, 14)
-
         SCAN_RANDOM = 0x01
         OWN_TYPE = SCAN_RANDOM
         SCAN_TYPE = 0x01
 
 
-    def parse_events(self, sock, loop_count=100):
+    def _parse_events(self, sock, loop_count=100):
+        """parse ble device output, filter and return ble advertisments"""
         old_filter = sock.getsockopt(bluez.SOL_HCI, bluez.HCI_FILTER, 14)
-
         # perform a device inquiry on bluetooth device #0
         # The inquiry should last 8 * 1.28 = 10.24 seconds
         # before the inquiry is performed, bluez should flush its cache of
@@ -122,14 +82,10 @@ def parse_events(self, sock, loop_count=100):
         bluez.hci_filter_all_events(flt)
         bluez.hci_filter_set_ptype(flt, bluez.HCI_EVENT_PKT)
         sock.setsockopt(bluez.SOL_HCI, bluez.HCI_FILTER, flt)
-        done = False
         results = []
-        myFullList = []
         for i in range(0, loop_count):
             pkt = sock.recv(255)
-
             ptype, event, plen = struct.unpack("BBB", pkt[:3])
-            # print( "--------------" )
             if event == bluez.EVT_INQUIRY_RESULT_WITH_RSSI:
                 i = 0
             elif event == bluez.EVT_NUM_COMP_PKTS:
@@ -142,20 +98,17 @@ def parse_events(self, sock, loop_count=100):
                 if subevent == self.EVT_LE_CONN_COMPLETE:
                     self.le_handle_connection_complete(pkt)
                 elif subevent == self.EVT_LE_ADVERTISING_REPORT:
-                    # print( "advertising report")
                     num_reports =  pkt[0]
                     report_pkt_offset = 0
                     for i in range(0, num_reports):
-                        myFullList.append({"data": pkt.hex(), "ts": str(datetime.datetime.utcnow())})
-
-                    done = True
+                        results.append({"data": pkt.hex(), "ts": str(datetime.datetime.utcnow())})
         sock.setsockopt(bluez.SOL_HCI, bluez.HCI_FILTER, old_filter)
-        #print(myFullList)
-        return myFullList
+        return results
 
 
     def scan(self,numberOfScannes=1):
-        return self.parse_events(self.sock, numberOfScannes)
+        """scan ble advertisment and return selected number of them"""
+        return self._parse_events(self.sock, numberOfScannes)
 
 #testing the helper class and write results to ble.txt
 if __name__ == '__main__':

bleScanner/HTTPPostHelper.py

@@ -14,33 +14,37 @@ def __init__(self, Id, Name,PostAddress):
         self.sended = False
         self.beaconData_ = []
 
-    def postData(self):
-        #generating needed json structure
-
+    def _cache_data(self, BeaconData){
+        """Update cache."""
+        # check if the data is posted before flushing the cache
+        # if data is not postet it will append new data to the old data
+        if self.sended:
+            self.beaconData_.clear()
+            self.beaconData_.extend(BeaconData)
+        else:
+            self.beaconData_.extend(BeaconData)
+    }
 
-        data = {
+    def _package_data(self)
+        """make post package."""
+        return {
             'id': self.Id,
             'name': self.Name,
             'mac' : self.mac,
             'ts' : str(datetime.datetime.utcnow()),
-            'data' : self.beaconData_
+            'data' : str(self.beaconData_)
         }
-        #print(json.dumps(data))
+
+    def _post_data(self):
+        """send packaged data to server via http post."""
+        data = _package_data()
         response = requests.post(self.PostAddress, json=json.dumps(data))
         if response.status_code == 200 :
             self.sended = True
         else:
             self.sended = False
 
-
-    def startPosting(self,BeaconData):
-        # check if the data is posted before flushing it
-        # if data is not postet it will append new data to the list
-        if self.sended:
-        #    print('Posted')
-            self.beaconData_.clear()
-            self.beaconData_.extend(BeaconData)
-        else:
-            self.beaconData_.extend(BeaconData)
-
-        self.postData()
+    def start_posting(self,BeaconData):
+        """cache and send scan data to server"""
+        self._cache_data(BeaconData)
+        self._post_data()

bleScanner/bleScanner.py

@@ -14,35 +14,39 @@
 postAddress = config['serverConfigurations']['postAddress']
 id = config['GatewayConfiguration']['id']
 name = config['GatewayConfiguration']['name']
-interval = int(config['GatewayConfiguration']['PostInterval'])
+postInterval = int(config['GatewayConfiguration']['PostInterval'])
 numScans = int(config['scannerConfiguration']['scanBuffer'])
 # shared list used to exchange Beacon scans with post thread
 returnedList =[]
 
 ScanThreat = BleHelper.BleHelper(0)
 HttpPostThreat = HTTPPostHelper.HTTPPostHelper(id, name,postAddress)
 
-def scanProcess():
+def scan_process():
     while True:
         returnedList.extend(ScanThreat.scan(numScans))
 
-def postProcess():
+def post_process():
     while True:
-        #temporary list used for exchange
+        #temporary list used to cache beacon data before posting
         templist = returnedList.copy()
         returnedList.clear()
         try:
-            HttpPostThreat.startPosting(templist)
+            HttpPostThreat.start_posting(templist)
         except:
-            print('couldnt post data')
-        time.sleep(interval)
+            print("couldn't post data")
+        time.sleep(postInterval)
 
 
 try:
-   _thread.start_new_thread(scanProcess, ())
-   _thread.start_new_thread(postProcess, ())
-except:
-   print ("Error: unable to start thread")
+   _thread.start_new_thread(scan_process, ())
+   _thread.start_new_thread(post_process, ())
+except Exception as e:
+   print("Error: unable to start thread")
+   print(e)
+   sys.exit(1)
+
 
 while True:
+    """keep the program running"""
    pass

btScanner/HTTPPostHelper.py

@@ -14,30 +14,37 @@ def __init__(self, Id, Name,PostAddress):
         self.sended = False
         self.beaconData_ = []
 
-    def postData(self):
-        #generating needed json structure
-        data = {
+    def _cache_data(self, BeaconData){
+        """Update cache."""
+        # check if the data is posted before flushing the cache
+        # if data is not postet it will append new data to the old data
+        if self.sended:
+            self.beaconData_.clear()
+            self.beaconData_.extend(BeaconData)
+        else:
+            self.beaconData_.extend(BeaconData)
+    }
+
+    def _package_data(self)
+        """make post package."""
+        return {
             'id': self.Id,
             'name': self.Name,
             'mac' : self.mac,
             'ts' : str(datetime.datetime.utcnow()),
             'data' : str(self.beaconData_)
         }
-        #print(json.dumps(data))
+
+    def _post_data(self):
+        """send packaged data to server via http post."""
+        data = _package_data()
         response = requests.post(self.PostAddress, json=json.dumps(data))
         if response.status_code == 200 :
             self.sended = True
         else:
             self.sended = False
 
-    def startPosting(self,BeaconData):
-        # check if the data is posted before flushing it
-        # if data is not postet it will append new data to the list
-        if self.sended:
-        #    print('Posted')
-            self.beaconData_.clear()
-            self.beaconData_.extend(BeaconData)
-        else:
-            self.beaconData_.extend(BeaconData)
-
-        self.postData()
+    def start_posting(self,BeaconData):
+        """cache and send scan data to server"""
+        self._cache_data(BeaconData)
+        self._post_data()