OfflineFindRecovery/src/python/findmy-discover.py at master · hajekj/OfflineFindRecovery · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
import asyncio
import logging
from datetime import datetime
import csv
import os

from scanner import OfflineFindingScanner

logging.basicConfig(level=logging.INFO)

pubkeys = []

def calculate_ble_distance(RSSI_at1m, RSSI, n):
    """
    Calculate the distance between a BLE transmitter and receiver using RSSI.

    Parameters:
    - RSSI_at1m: The RSSI value measured at 1 meter from the transmitter.
    - RSSI: The measured RSSI value at the receiver.
    - n: The path loss exponent.

    Returns:
    - The estimated distance in meters.
    """
    distance = 10 ** ((RSSI_at1m - RSSI) / (10 * n))
    return distance

class CSVWriter():

    filename = None
    fp = None
    writer = None

    def __init__(self, filename):
        self.filename = filename
        self.fp = open(self.filename, 'w', encoding='utf8')
        self.writer = csv.writer(self.fp, delimiter=';', quotechar='"', quoting=csv.QUOTE_ALL, lineterminator='\n')

    def close(self):
        self.fp.close()

    def write(self, *args):
        self.writer.writerow(args)

    def size(self):
        return os.path.getsize(self.filename)

    def fname(self):
        return self.filename

async def scan() -> None:
    scanner = await OfflineFindingScanner.create()

    file = open('discovery-keys.csv', "r")
    csvreader = csv.reader(file, delimiter=";", quotechar='"', quoting=csv.QUOTE_ALL, lineterminator='\n')
    for row in csvreader:
        pubkeys.append(row[1])
    file.close()

    print("Scanning for FindMy-devices...")
    print()
    mycsv = CSVWriter('discovery-output.csv')

    RSSI_at1m = -52
    pathLossExponent = 2.0

    async for device in scanner.scan_for(2 * 60, extend_timeout=True):
        if device.status == 0:
            distance = calculate_ble_distance(RSSI_at1m, device.additional_data.get('rssi'), pathLossExponent)
            mycsv.write(device.mac_address,str(datetime.utcnow()),device.adv_key_b64,device.additional_data.get('rssi'),distance)

            if  device.adv_key_b64 in pubkeys:
                distanceStr = "Distance"
                print(f"Device - {device.mac_address}")
                print(f"  Time:         {datetime.utcnow()}")
                print(f"  Public key:   {device.adv_key_b64}")
                print(f"  Lookup key:   {device.hashed_adv_key_b64}")
                print(f"  Status byte:  {device.status:x}")
                print(f"  Hint byte:    {device.hint:x}")
                print("  Extra data:")
                print(f"    {distanceStr:20}: {distance}")
                for k, v in sorted(device.additional_data.items()):
                    print(f"    {k:20}: {v}")
                print()
            # else:
            #     print(f"Device - {device.mac_address}")
            #     print(f"  Time:         {datetime.utcnow()}")
            #     print(f"  Public key:   {device.adv_key_b64}")
            #     for k, v in sorted(device.additional_data.items()):
            #         print(f"    {k:20}: {v}")
        # else:
        #     print(f"Device - {device.mac_address}")
        #     print(f"  Time:         {datetime.utcnow()}")
        #     print(f"  Public key:   {device.adv_key_b64}")
        #     print(f"  Status byte:  {device.status:x}")
        #     for k, v in sorted(device.additional_data.items()):
        #         print(f"    {k:20}: {v}")

    mycsv.close()
    print("Closing")

if __name__ == "__main__":
    asyncio.run(scan())