set_rtc_time_precise.py

#!/usr/bin/env python3
"""
Precise DS3231 RTC time synchronization with automatic calibration.
Provides sub-second accuracy by measuring and compensating for serial delays.
"""

import serial
import serial.tools.list_ports
import time
from datetime import datetime, timedelta
import sys
import argparse
import statistics

def find_ch32_port():
    """Find the CH32V003 serial port automatically."""
    ports = serial.tools.list_ports.comports()

    # Common CH32/WCH Link device identifiers
    ch32_identifiers = ['1a86:8010', 'WCH', 'CH32', 'USB Serial', '/dev/tty.usbserial', '/dev/ttyUSB', '/dev/ttyACM']

    for port in ports:
        port_str = str(port).lower()
        device_str = port.device.lower()

        # Check various identifiers
        for identifier in ch32_identifiers:
            if identifier.lower() in port_str or identifier.lower() in device_str:
                return port.device

        # Also check VID:PID
        if hasattr(port, 'vid') and hasattr(port, 'pid'):
            if port.vid == 0x1a86:  # WCH vendor ID
                return port.device

    # If not found, list available ports for manual selection
    if ports:
        print("Available serial ports:")
        for i, port in enumerate(ports):
            print(f"  {i}: {port.device} - {port.description}")

        try:
            choice = int(input("Select port number: "))
            if 0 <= choice < len(ports):
                return ports[choice].device
        except (ValueError, IndexError):
            pass

    return None

def measure_serial_latency(ser, samples=5):
    """Measure average serial round-trip latency."""
    latencies = []

    for i in range(samples):
        # Clear buffers
        ser.reset_input_buffer()
        ser.reset_output_buffer()

        # Measure round-trip time for menu command
        start = time.perf_counter()
        ser.write(b'h')
        time.sleep(0.01)  # Small delay to ensure transmission

        # Wait for response (up to 1 second)
        timeout = time.time() + 1
        while ser.in_waiting == 0 and time.time() < timeout:
            time.sleep(0.001)

        if ser.in_waiting:
            # Read response
            response = ser.read(ser.in_waiting)
            end = time.perf_counter()

            # Calculate one-way latency (half of round-trip)
            latency = (end - start) / 2
            latencies.append(latency * 1000)  # Convert to ms

            # Small delay between measurements
            time.sleep(0.1)

    if latencies:
        avg_latency = statistics.mean(latencies)
        std_latency = statistics.stdev(latencies) if len(latencies) > 1 else 0
        return avg_latency, std_latency

    return 100, 50  # Default fallback values in ms

def sync_to_next_second():
    """Synchronize to the next second boundary with microsecond precision."""
    now = datetime.now()
    microseconds_remaining = 1000000 - now.microsecond

    # Wait until we're very close to the next second
    time.sleep(microseconds_remaining / 1000000.0 - 0.001)  # Stop 1ms early

    # Busy-wait for the final millisecond for better precision
    target_time = now.replace(microsecond=0) + timedelta(seconds=1)
    while datetime.now() < target_time:
        pass

    return datetime.now()

def set_rtc_time_precise(port, baud_rate=115200, timezone_offset=None, auto_calibrate=True, compensation_ms=None):
    """Set the RTC time with high precision using automatic calibration.

    Args:
        port: Serial port path
        baud_rate: Communication speed (default 115200)
        timezone_offset: Hours offset from local time
        auto_calibrate: Automatically measure and compensate for serial latency
        compensation_ms: Manual compensation in milliseconds (overrides auto-calibration)
    """

    try:
        print(f"Connecting to {port} at {baud_rate} baud...")
        ser = serial.Serial(port, baud_rate, timeout=1)
        time.sleep(2)  # Wait for device to initialize

        # Measure serial latency if auto-calibrating
        if auto_calibrate and compensation_ms is None:
            print("Measuring serial communication latency...")
            avg_latency, std_latency = measure_serial_latency(ser, samples=5)

            # Add processing time estimate (based on typical MCU response time)
            processing_time = 50  # ms for MCU to process and set RTC

            # Calculate total compensation
            # Add 2x standard deviation for safety margin
            compensation_ms = avg_latency + processing_time + (2 * std_latency)

            print(f"  Average latency: {avg_latency:.1f}ms (±{std_latency:.1f}ms)")
            print(f"  Using total compensation: {compensation_ms:.1f}ms")
        elif compensation_ms is None:
            compensation_ms = 200  # Default fallback
            print(f"  Using default compensation: {compensation_ms}ms")
        else:
            print(f"  Using manual compensation: {compensation_ms}ms")

        # Clear any pending data
        ser.reset_input_buffer()
        ser.reset_output_buffer()

        # Send 'h' to get menu first
        ser.write(b'h')
        time.sleep(0.5)

        # Read and display initial response
        if ser.in_waiting:
            response = ser.read(ser.in_waiting).decode('utf-8', errors='ignore')
            print("\nDevice menu ready")

        # Clear buffer again
        ser.reset_input_buffer()

        print("\nSynchronizing to next second boundary...")

        # Prepare command in advance
        time_command = b't'  # Letter command for set time

        # Wait for next second boundary
        sync_time = sync_to_next_second()

        # Calculate the target time with compensation
        target_time = sync_time + timedelta(milliseconds=compensation_ms)
        if timezone_offset:
            target_time = target_time + timedelta(hours=timezone_offset)

        # Send time setting command immediately after sync
        ser.write(time_command)
        time.sleep(0.15)  # Minimal delay for command processing

        # Send the time string
        time_str = target_time.strftime("%H:%M:%S")
        date_str = target_time.strftime("%Y-%m-%d")

        print(f"Setting time to: {target_time.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}")
        print(f"  System time at sync: {sync_time.strftime('%H:%M:%S.%f')[:-3]}")
        print(f"  Compensation applied: {compensation_ms:.1f}ms")

        # Send time
        ser.write(time_str.encode())
        ser.write(b'\r')
        time.sleep(0.2)

        # Read response
        response = ser.read(1000).decode('utf-8', errors='ignore')
        if "OK" in response or "successfully" in response.lower():
            print("  ✓ Time set successfully")

        # Set date
        time.sleep(0.2)
        ser.write(b'd')  # Select "Set date" option (letter command)
        time.sleep(0.15)

        # Send date
        ser.write(date_str.encode())
        ser.write(b'\r')
        time.sleep(0.2)

        # Read response
        response = ser.read(1000).decode('utf-8', errors='ignore')
        if "OK" in response or "successfully" in response.lower():
            print("  ✓ Date set successfully")
            # The C code now automatically calculates the correct day-of-week
            # when setting the date, so no additional action needed

        # Verify accuracy by reading back immediately
        print("\nVerifying synchronization accuracy...")
        time.sleep(0.3)

        # Read current RTC time multiple times to check consistency
        readings = []
        for i in range(3):
            ser.write(b'r')  # Read time command (letter command)
            time.sleep(0.15)
            read_time = datetime.now()

            if ser.in_waiting:
                response = ser.read(ser.in_waiting).decode('utf-8', errors='ignore')
                lines = response.strip().split('\n')

                for line in lines:
                    # Parse ISO 8601 format: YYYY-MM-DDTHH:MM:SS Day
                    if 'T' in line and ':' in line:
                        try:
                            # Extract just the datetime part (before day name)
                            datetime_str = line.strip().split()[0] if ' ' in line else line.strip()

                            if i == 0:  # Only print first reading
                                print(f"  RTC time: {datetime_str}")
                                print(f"  System time: {read_time.strftime('%Y-%m-%dT%H:%M:%S')}")

                            # Parse ISO 8601 format
                            if 'T' in datetime_str:
                                time_part = datetime_str.split('T')[1]
                                rtc_h, rtc_m, rtc_s = map(int, time_part.split(':'))
                                rtc_seconds = rtc_h * 3600 + rtc_m * 60 + rtc_s

                                sys_seconds = read_time.hour * 3600 + read_time.minute * 60 + read_time.second

                                # Calculate difference
                                diff = rtc_seconds - sys_seconds

                                # Handle day boundary
                                if diff > 43200:  # More than 12 hours ahead
                                    diff -= 86400
                                elif diff < -43200:  # More than 12 hours behind
                                    diff += 86400

                                readings.append(diff)
                        except Exception as e:
                            pass

            time.sleep(0.5)

        if readings:
            avg_diff = statistics.mean(readings)
            print(f"\n  Average time difference: {avg_diff:+.1f} seconds")

            if abs(avg_diff) < 0.5:
                print("  ✓ Excellent synchronization achieved!")
            elif abs(avg_diff) < 1.0:
                print("  ✓ Good synchronization achieved")
                print(f"  Tip: Try increasing compensation by {int(avg_diff * 1000)}ms for better accuracy")
            else:
                print(f"  ⚠ RTC is off by {avg_diff:.1f} seconds")
                suggested_compensation = compensation_ms + (avg_diff * 1000)
                print(f"  Tip: Try -c {int(suggested_compensation)} for better accuracy")

        print("\n✓ RTC time synchronization complete!")

        ser.close()
        return True

    except serial.SerialException as e:
        print(f"Serial error: {e}")
        return False
    except KeyboardInterrupt:
        print("\nInterrupted by user")
        return False
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
        return False

def main():
    parser = argparse.ArgumentParser(
        description='Precise DS3231 RTC time synchronization with automatic calibration',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s                           # Auto-detect port and auto-calibrate
  %(prog)s -p /dev/ttyUSB0          # Specific port with auto-calibration
  %(prog)s -c 500                   # Manual 500ms compensation
  %(prog)s --no-calibrate           # Skip auto-calibration, use default

Tips for best accuracy:
  - Run multiple times to find the optimal compensation value
  - Use a stable USB connection (avoid hubs if possible)
  - Close other applications that might cause system delays
        """
    )

    parser.add_argument('-p', '--port', help='Serial port (auto-detect if not specified)')
    parser.add_argument('-b', '--baud', type=int, default=115200, help='Baud rate (default: 115200)')
    parser.add_argument('-t', '--timezone', type=float, help='Timezone offset in hours (e.g., -5 for EST)')
    parser.add_argument('-c', '--compensation', type=int,
                        help='Manual compensation in milliseconds (overrides auto-calibration)')
    parser.add_argument('--no-calibrate', action='store_true',
                        help='Skip auto-calibration and use default compensation')

    args = parser.parse_args()

    # Find serial port
    port = args.port
    if not port:
        print("Auto-detecting CH32V003 serial port...")
        port = find_ch32_port()

        if not port:
            print("Error: Could not find CH32V003 device.")
            print("Please specify the port with -p option")
            sys.exit(1)

    print(f"Using port: {port}")

    # Determine calibration mode
    auto_calibrate = not args.no_calibrate and args.compensation is None

    # Set the RTC time with precision
    if set_rtc_time_precise(port, args.baud, args.timezone, auto_calibrate, args.compensation):
        sys.exit(0)
    else:
        sys.exit(1)

if __name__ == "__main__":
    main()