GoogleFindMyTools for raspberry over Mqqt to HA

Auth/fcm_receiver.py

@@ -1,116 +1,131 @@
 import asyncio
 import base64
 import binascii
-
 from Auth.firebase_messaging import FcmRegisterConfig, FcmPushClient
 from Auth.token_cache import set_cached_value, get_cached_value
 
 class FcmReceiver:
-
     _instance = None
     _listening = False
-
+    
     def __new__(cls, *args, **kwargs):
         if cls._instance is None:
-            cls._instance = super(FcmReceiver, cls).__new__(cls, *args, **kwargs)
+            cls._instance = super(FcmReceiver, cls).__new__(cls)
         return cls._instance
-
+    
     def __init__(self):
         if hasattr(self, '_initialized') and self._initialized:
             return
         self._initialized = True
-
+        
         # Define Firebase project configuration
         project_id = "google.com:api-project-289722593072"
         app_id = "1:289722593072:android:3cfcf5bc359f0308"
         api_key = "AIzaSyD_gko3P392v6how2H7UpdeXQ0v2HLettc"
         message_sender_id = "289722593072"
-
         fcm_config = FcmRegisterConfig(
             project_id=project_id,
             app_id=app_id,
             api_key=api_key,
             messaging_sender_id=message_sender_id,
             bundle_id="com.google.android.apps.adm",
         )
-
         self.credentials = get_cached_value('fcm_credentials')
         self.location_update_callbacks = []
         self.pc = FcmPushClient(self._on_notification, fcm_config, self.credentials, self._on_credentials_updated)
-
-
-    def register_for_location_updates(self, callback):
-
+        self.listen_task = None
+        self.timeout_task = None
+
+    def register_for_location_updates(self, callback, timeout_seconds=60):
         if not self._listening:
-            asyncio.get_event_loop().run_until_complete(self._register_for_fcm_and_listen())
-
+            asyncio.get_event_loop().run_until_complete(
+                self._register_for_fcm_and_listen(timeout_seconds)
+            )
         self.location_update_callbacks.append(callback)
-
         return self.credentials['fcm']['registration']['token']
-
-
+
     def stop_listening(self):
+        if self.timeout_task and not self.timeout_task.done():
+            self.timeout_task.cancel()
+        if self.listen_task and not self.listen_task.done():
+            self.listen_task.cancel()
         asyncio.get_event_loop().run_until_complete(self.pc.stop())
         self._listening = False
-
-
+
     def get_android_id(self):
-
         if self.credentials is None:
-            return asyncio.get_event_loop().run_until_complete(self._register_for_fcm_and_listen())
-
+            return asyncio.get_event_loop().run_until_complete(
+                self._register_for_fcm_and_listen()
+            )
         return self.credentials['gcm']['android_id']
-
-
+
     # Define a callback function for handling notifications
     def _on_notification(self, obj, notification, data_message):
-
+        # Reset the timeout timer when we receive a notification
+        if self.timeout_task and not self.timeout_task.done():
+            self.timeout_task.cancel()
+
         # Check if the payload is present
         if 'data' in obj and 'com.google.android.apps.adm.FCM_PAYLOAD' in obj['data']:
-
             # Decode the base64 string
             base64_string = obj['data']['com.google.android.apps.adm.FCM_PAYLOAD']
             decoded_bytes = base64.b64decode(base64_string)
-
-            # print("[FCMReceiver] Decoded FMDN Message:", decoded_bytes.hex())
-
             # Convert to hex string
             hex_string = binascii.hexlify(decoded_bytes).decode('utf-8')
-
             for callback in self.location_update_callbacks:
                 callback(hex_string)
         else:
             print("[FCMReceiver] Payload not found in the notification.")
-
-
+
     def _on_credentials_updated(self, creds):
         self.credentials = creds
-
         # Also store to disk
         set_cached_value('fcm_credentials', self.credentials)
         print("[FCMReceiver] Credentials updated.")
-
-
+
+    async def _timeout_handler(self, timeout_seconds):
+        try:
+            await asyncio.sleep(timeout_seconds)
+            print(f"[FCMReceiver] Timed out after {timeout_seconds} seconds")
+            if self._listening:
+                await self.pc.stop()
+                self._listening = False
+        except asyncio.CancelledError:
+            # This is normal when a notification is received and the timeout is canceled
+            pass
+
     async def _register_for_fcm(self):
         fcm_token = None
-
         # Register or check in with FCM and get the FCM token
         while fcm_token is None:
             try:
                 fcm_token = await self.pc.checkin_or_register()
             except Exception as e:
                 await self.pc.stop()
-                print("[FCMReceiver] Failed to register with FCM. Retrying...")
+                print(f"[FCMReceiver] Failed to register with FCM: {str(e)}. Retrying...")
                 await asyncio.sleep(5)
-
-
-    async def _register_for_fcm_and_listen(self):
+
+    async def _register_for_fcm_and_listen(self, timeout_seconds=60):
         await self._register_for_fcm()
-        await self.pc.start()
+
+        self.listen_task = asyncio.create_task(self.pc.start())
         self._listening = True
         print("[FCMReceiver] Listening for notifications. This can take a few seconds...")
-
+
+        # Set up the timeout
+        if timeout_seconds > 0:
+            self.timeout_task = asyncio.create_task(self._timeout_handler(timeout_seconds))
 
 if __name__ == "__main__":
     receiver = FcmReceiver()
-    print(receiver.get_android_id())
+    try:
+        # Example usage with a 30-second timeout
+        def on_location_update(hex_data):
+            print(f"Received location update: {hex_data[:20]}...")
+
+        receiver.register_for_location_updates(on_location_update, timeout_seconds=30)
+        # Keep the main thread running
+        asyncio.get_event_loop().run_forever()
+    except KeyboardInterrupt:
+        print("Stopping...")
+        receiver.stop_listening()

NovaApi/ExecuteAction/LocateTracker/decrypt_locations.py

@@ -129,32 +129,47 @@ def decrypt_location_response_locations(device_update_protobuf):
 
     if not location_time_array:
         print("No locations found.")
-        return
-
-    for loc in location_time_array:
-
-        if loc.status == Common_pb2.Status.SEMANTIC:
-            print(f"Semantic Location: {loc.name}")
-
-        else:
-            proto_loc = DeviceUpdate_pb2.Location()
-            proto_loc.ParseFromString(loc.decrypted_location)
-
-            latitude = proto_loc.latitude / 1e7
-            longitude = proto_loc.longitude / 1e7
-            altitude = proto_loc.altitude
-
-            print(f"Latitude: {latitude}")
-            print(f"Longitude: {longitude}")
-            print(f"Altitude: {altitude}")
-            print(f"Google Maps Link: {create_google_maps_link(latitude, longitude)}")
-
+        return None
+
+    # Return data from the most recent location
+    loc = location_time_array[0]
+
+    if loc.status == Common_pb2.Status.SEMANTIC:
+        print(f"Semantic Location: {loc.name}")
+        location_data = {
+            'semantic_location': loc.name,
+            'timestamp': datetime.datetime.fromtimestamp(loc.time).strftime('%Y-%m-%d %H:%M:%S'),
+            'status': loc.status,
+            'is_own_report': loc.is_own_report
+        }
+    else:
+        proto_loc = DeviceUpdate_pb2.Location()
+        proto_loc.ParseFromString(loc.decrypted_location)
+
+        latitude = proto_loc.latitude / 1e7
+        longitude = proto_loc.longitude / 1e7
+        altitude = proto_loc.altitude
+
+        print(f"Latitude: {latitude}")
+        print(f"Longitude: {longitude}")
+        print(f"Altitude: {altitude}")
+        print(f"Google Maps Link: {create_google_maps_link(latitude, longitude)}")
         print(f"Time: {datetime.datetime.fromtimestamp(loc.time).strftime('%Y-%m-%d %H:%M:%S')}")
         print(f"Status: {loc.status}")
         print(f"Is Own Report: {loc.is_own_report}")
         print("-" * 40)
 
-    pass
+        location_data = {
+            'latitude': latitude,
+            'longitude': longitude,
+            'altitude': altitude,
+            'accuracy': loc.accuracy,
+            'timestamp': datetime.datetime.fromtimestamp(loc.time).strftime('%Y-%m-%d %H:%M:%S'),
+            'status': loc.status,
+            'is_own_report': loc.is_own_report
+        }
+
+    return location_data
 
 
 if __name__ == '__main__':

NovaApi/ExecuteAction/LocateTracker/location_request.py

@@ -53,7 +53,8 @@ def handle_location_response(response):
     while result is None:
         asyncio.get_event_loop().run_until_complete(asyncio.sleep(0.1))
 
-    decrypt_location_response_locations(result)
+    locations = decrypt_location_response_locations(result)
+    return locations
 
 if __name__ == '__main__':
     get_location_data_for_device(get_example_data("sample_canonic_device_id"), "Test")

NovaApi/ListDevices/nbe_list_devices.py

@@ -1,8 +1,3 @@
-#
-#  GoogleFindMyTools - A set of tools to interact with the Google Find My API
-#  Copyright © 2024 Leon Böttger. All rights reserved.
-#
-
 import binascii
 from NovaApi.ExecuteAction.LocateTracker.location_request import get_location_data_for_device
 from NovaApi.nova_request import nova_request
@@ -15,10 +10,8 @@
 
 
 def request_device_list():
-
     hex_payload = create_device_list_request()
     result = nova_request(NOVA_LIST_DEVICS_API_SCOPE, hex_payload)
-
     return result
 
 
@@ -56,20 +49,12 @@ def list_devices():
     print("")
     print("The following trackers are available:")
 
+    # Anstatt Eingabeaufforderung, jedes Gerät automatisch durchlaufen
     for idx, (device_name, canonic_id) in enumerate(canonic_ids, start=1):
         print(f"{idx}. {device_name}: {canonic_id}")
 
-    selected_value = input("\nIf you want to see locations of a tracker, type the number of the tracker and press 'Enter'.\nIf you want to register a new ESP32- or Zephyr-based tracker, type 'r' and press 'Enter': ")
-
-    if selected_value == 'r':
-        print("Loading...")
-        register_esp32()
-    else:
-        selected_idx = int(selected_value) - 1
-        selected_device_name = canonic_ids[selected_idx][0]
-        selected_canonic_id = canonic_ids[selected_idx][1]
-
-        get_location_data_for_device(selected_canonic_id, selected_device_name)
+        # Automatisch den Standort jedes Geräts abfragen
+        get_location_data_for_device(canonic_id, device_name)
 
 
 if __name__ == '__main__':