Auto Time Sync script

This program automatically synchronizes the time on BTHome-compatible Bluetooth Low Energy (BLE) THB2 firmware flashed devices. It continuously scans for devices that advertise the BTHome service UUID (0xFCD2) and ensures their internal clocks are synchronized with the host computer's time. This script could be executed by a scheduled cron job for instance to rectify the time drift of the devices. Key Features: - Discovers BTHome devices using service UUID 0xFCD2 - Automatically connects and syncs time when drift exceeds 3 seconds - Supports multiple devices with concurrent processing - Implements retry mechanism for failed connections - Prevents excessive reconnections with 60-second cooldown - Includes timeout protection for unresponsive devices Technical Implementation: 1. Device Discovery: - Uses BleakScanner with a detection callback - Filters devices by BTHome service UUID (0xFCD2) - Checks both service UUIDs and service data 2. Connection Process: - Establishes BLE connection with 30-second timeout - Discovers services and sets up notifications - Uses characteristic 0xFFF4 for time sync commands 3. Time Synchronization: - Reads current device time via command 0x33 - Compares with system time accounting for timezone - Updates device time if difference exceeds 3 seconds - Uses command 0x23 to set new time value 4. Error Handling: - Implements connection retries (max 2 attempts) - Includes timeout protection for all BLE operations - Gracefully handles disconnections and exceptions Usage: Run the script directly to start the auto time sync service: python auto_time_sync.py The program will continuously scan for and process BTHome devices, logging all operations and any errors that occur. Use Ctrl+C to gracefully stop the service.
2025-01-25 17:39:23 +00:00 · 2025-01-25 17:39:23 +00:00 · 9aedbf3bbc
commit 9aedbf3bbc
parent c3d63e24ab
2 changed files with 283 additions and 0 deletions
--- a/auto_time_sync.py
+++ b/auto_time_sync.py
@ -0,0 +1,270 @@
+"""
+Auto Time Sync for BTHome Devices
+--------------------------------
+
+This program automatically synchronizes the time on BTHome-compatible Bluetooth Low Energy (BLE) devices.
+It continuously scans for devices that advertise the BTHome service UUID (0xFCD2) and ensures their 
+internal clocks are synchronized with the host computer's time.
+
+Key Features:
+- Discovers BTHome devices using service UUID 0xFCD2
+- Automatically connects and syncs time when drift exceeds 3 seconds
+- Supports multiple devices with concurrent processing
+- Implements retry mechanism for failed connections
+- Prevents excessive reconnections with 60-second cooldown
+- Includes timeout protection for unresponsive devices
+
+Technical Implementation:
+1. Device Discovery:
+   - Uses BleakScanner with a detection callback
+   - Filters devices by BTHome service UUID (0xFCD2)
+   - Checks both service UUIDs and service data
+
+2. Connection Process:
+   - Establishes BLE connection with 30-second timeout
+   - Discovers services and sets up notifications
+   - Uses characteristic 0xFFF4 for time sync commands
+
+3. Time Synchronization:
+   - Reads current device time via command 0x33
+   - Compares with system time accounting for timezone
+   - Updates device time if difference exceeds 3 seconds
+   - Uses command 0x23 to set new time value
+
+4. Error Handling:
+   - Implements connection retries (max 2 attempts)
+   - Includes timeout protection for all BLE operations
+   - Gracefully handles disconnections and exceptions
+
+Usage:
+Run the script directly to start the auto time sync service:
+    python auto_time_sync.py
+
+The program will continuously scan for and process BTHome devices, logging all operations
+and any errors that occur. Use Ctrl+C to gracefully stop the service.
+"""
+
+import asyncio
+import time
+from datetime import datetime
+from bleak import BleakScanner, BleakClient
+from bleak.backends.device import BLEDevice
+
+class AutoTimeSync:
+    def __init__(self):
+        self.device = None
+        self.client = None
+        self.cmd_characteristic = "0000fff4-0000-1000-8000-00805f9b34fb"
+        self.service_uuid = "0000fcd2-0000-1000-8000-00805f9b34fb"
+        self.is_handling_notification = False
+        self.processed_devices = {}  # Track when each device was last processed
+        self.current_task = None
+        self.start_time = time.time()  # Track when the program started
+
+    def add_log(self, text: str, is_error: bool = False):
+        """Log messages with timestamp"""
+        timestamp = datetime.now().strftime("%H:%M:%S")
+        prefix = "ERROR: " if is_error else ""
+        print(f"{timestamp}: {prefix}{text}")
+
+    def set_status(self, text: str, is_error: bool = False):
+        """Update status with optional error indication"""
+        prefix = "ERROR: " if is_error else "Status: "
+        print(f"{prefix}{text}")
+
+    async def set_device_time(self):
+        """Set the device time"""
+        if not self.client or not self.client.is_connected:
+            return False
+
+        # Calculate current time in seconds since epoch, adjusted for timezone
+        current_time = int(time.time())
+        timezone_offset = time.timezone if (time.localtime().tm_isdst == 0) else time.altzone
+        adjusted_time = current_time - timezone_offset
+
+        # Create time sync packet
+        time_bytes = bytearray([0x23])  # Command byte
+        time_bytes.extend(adjusted_time.to_bytes(4, 'little'))  # Time in little endian
+
+        try:
+            await self.client.write_gatt_char(self.cmd_characteristic, time_bytes, response=True)
+            self.add_log('Time set successfully')
+            return True
+        except Exception as e:
+            self.add_log(f'Error setting time: {str(e)}', True)
+            return False
+
+    async def handle_notification(self, sender: int, data: bytearray):
+        """Handle incoming notifications from the device"""
+        if not data or self.is_handling_notification:
+            return
+
+        self.is_handling_notification = True
+        try:
+            blk_id = data[0]
+            
+            # Handle measurement data packet (0x33)
+            if blk_id == 0x33 and len(data) >= 15:
+                device_time = int.from_bytes(data[11:15], 'little')
+                local_time = int(time.time()) - time.timezone
+                time_diff = abs(device_time - local_time)
+                
+                self.add_log(f'Device time: {datetime.fromtimestamp(device_time)}')
+                self.add_log(f'Time difference: {time_diff} seconds')
+                
+                time_updated = False
+                if time_diff > 3:
+                    self.add_log('Time difference > 3s, updating device time...')
+                    time_updated = await self.set_device_time()
+
+                # Disconnect after processing
+                self.add_log(f'Time {"updated" if time_updated else "checked"}, disconnecting...')
+                if self.client and self.client.is_connected:
+                    try:
+                        await self.client.disconnect()
+                    except:
+                        pass  # Ignore disconnect errors
+
+                # Update last processed time for this device
+                if self.device:
+                    self.processed_devices[self.device.address] = time.time()
+
+        finally:
+            self.is_handling_notification = False
+
+    async def process_device(self, device: BLEDevice, max_retries=2):
+        """Process a single device with retries"""
+        for attempt in range(max_retries):
+            try:
+                self.device = device
+                device_name = self.device.name if self.device.name else self.device.address
+                self.set_status(f'Connecting to device {device_name} (attempt {attempt + 1}/{max_retries})...')
+                
+                # Create a timeout for the entire connection process
+                async def connect_with_timeout():
+                    self.client = BleakClient(device, timeout=20.0)
+                    await self.client.connect()
+
+                    self.set_status('Connected, discovering services...')
+                    await asyncio.sleep(1)  # Give device time to settle
+                    if not self.client.services:
+                        raise Exception("No services found")
+
+                    self.set_status('Setting up notifications...')
+                    await self.client.start_notify(
+                        self.cmd_characteristic,
+                        self.handle_notification
+                    )
+
+                    self.set_status(f'Connected to {device_name}')
+                    
+                    # Request initial measurement and wait for response
+                    await self.client.write_gatt_char(
+                        self.cmd_characteristic,
+                        bytearray([0x33]),
+                        response=True
+                    )
+                    # Give some time for notification to arrive
+                    await asyncio.sleep(2)
+
+                try:
+                    # Set a 30-second timeout for the entire connection process
+                    await asyncio.wait_for(connect_with_timeout(), timeout=30.0)
+                    # If we get here, we succeeded so break the retry loop
+                    break
+                except asyncio.TimeoutError:
+                    raise Exception("Connection process timed out")
+
+            except Exception as error:
+                device_name = device.name if device.name else device.address
+                self.set_status(f'Error with {device_name} (attempt {attempt + 1}/{max_retries}): {str(error)}', True)
+                self.add_log(str(error), True)
+                if self.client and self.client.is_connected:
+                    await self.client.disconnect()
+                if attempt < max_retries - 1:
+                    await asyncio.sleep(2)  # Wait before retry
+
+    async def connect(self):
+        """Scan and process all THB devices"""
+        try:
+            self.set_status('Scanning for devices...')
+            
+            # Scan for devices with longer timeout
+            discovered_devices = []
+            def detection_callback(device, advertisement_data):
+                service_uuid_lower = self.service_uuid.lower()
+                # Check both service UUIDs and service data
+                has_service_uuid = (advertisement_data.service_uuids and 
+                                  service_uuid_lower in [str(uuid).lower() for uuid in advertisement_data.service_uuids])
+                has_service_data = service_uuid_lower in [str(uuid).lower() for uuid in advertisement_data.service_data.keys()]
+                
+                if has_service_uuid or has_service_data:
+                    discovered_devices.append(device)
+
+            scanner = BleakScanner(detection_callback=detection_callback)
+            await scanner.start()
+            await asyncio.sleep(15.0)  # Scan for 15 seconds
+            await scanner.stop()
+            
+            matching_devices = discovered_devices
+            
+            if not matching_devices:
+                self.add_log('No devices with BTHome service found', True)
+                return
+
+            current_time = time.time()
+            # Process each discovered device that hasn't been processed recently
+            for device in matching_devices:
+                # Skip devices processed in the last 60 seconds
+                if device.address in self.processed_devices:
+                    last_processed = self.processed_devices[device.address]
+                    if current_time - last_processed < 60:
+                        device_name = device.name if device.name else device.address
+                        self.add_log(f'Skipping {device_name} - processed recently')
+                        continue
+                
+                await self.process_device(device)
+
+        except Exception as error:
+            self.set_status(f'Error: {str(error)}', True)
+            self.add_log(str(error), True)
+
+    async def run(self):
+        """Main run loop"""
+        while time.time() - self.start_time < 600:  # Run for 10 minutes (600 seconds)
+            try:
+                await self.connect()
+                # Wait before next scan
+                await asyncio.sleep(1)
+            except asyncio.CancelledError:
+                self.add_log("Shutting down...")
+                if self.client and self.client.is_connected:
+                    await self.client.disconnect()
+                break
+            except Exception as e:
+                self.add_log(f"Error in main loop: {e}", True)
+                await asyncio.sleep(1)
+        
+        self.add_log("10 minute runtime completed")
+
+if __name__ == "__main__":
+    print("Auto Time Sync")
+    print("-------------")
+    print("Program will run for 10 minutes")
+    
+    sync = AutoTimeSync()
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    
+    try:
+        loop.run_until_complete(sync.run())
+    except KeyboardInterrupt:
+        print("\nShutting down due to user interrupt")
+    finally:
+        pending = asyncio.all_tasks(loop)
+        for task in pending:
+            task.cancel()
+        
+        # Give tasks a chance to complete
+        loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+        loop.close()