You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
132 lines
3.6 KiB
132 lines
3.6 KiB
#!/usr/bin/env python3
|
|
"""
|
|
Global Clock - Timestamp Synchronization
|
|
=========================================
|
|
Singleton clock for synchronizing timestamps across UART, I2C, and other modules.
|
|
|
|
All timestamps use the same time reference (app start), enabling:
|
|
- Aligned plotting of UART and I2C data
|
|
- Consistent timeline across all modules
|
|
- Database storage with unified timestamp column
|
|
|
|
Usage:
|
|
from global_clock import GlobalClock
|
|
|
|
clock = GlobalClock.instance()
|
|
timestamp = clock.now() # Seconds since app start
|
|
|
|
Author: Kynsight
|
|
Version: 1.0.0
|
|
"""
|
|
|
|
import time
|
|
from typing import Optional
|
|
|
|
|
|
class GlobalClock:
|
|
"""
|
|
Singleton clock for application-wide timestamp synchronization.
|
|
|
|
Uses monotonic time (time.perf_counter) for stability.
|
|
All timestamps are relative to application start time.
|
|
"""
|
|
|
|
_instance: Optional['GlobalClock'] = None
|
|
|
|
def __init__(self):
|
|
"""Initialize clock at application start."""
|
|
self._start_time = time.perf_counter()
|
|
|
|
@classmethod
|
|
def instance(cls) -> 'GlobalClock':
|
|
"""Get singleton instance (creates if needed)."""
|
|
if cls._instance is None:
|
|
cls._instance = GlobalClock()
|
|
return cls._instance
|
|
|
|
def now(self) -> float:
|
|
"""
|
|
Get current timestamp in seconds since app start.
|
|
|
|
Returns:
|
|
float: Time in seconds (e.g., 123.456789)
|
|
"""
|
|
return time.perf_counter() - self._start_time
|
|
|
|
def now_ns(self) -> int:
|
|
"""
|
|
Get current timestamp in nanoseconds since app start.
|
|
|
|
Returns:
|
|
int: Time in nanoseconds
|
|
"""
|
|
return int((time.perf_counter() - self._start_time) * 1_000_000_000)
|
|
|
|
def reset(self):
|
|
"""Reset clock to zero (useful for testing or new sessions)."""
|
|
self._start_time = time.perf_counter()
|
|
|
|
|
|
# =============================================================================
|
|
# Module-level convenience functions
|
|
# =============================================================================
|
|
|
|
def now() -> float:
|
|
"""
|
|
Get current timestamp (seconds since app start).
|
|
|
|
Convenience function for GlobalClock.instance().now()
|
|
"""
|
|
return GlobalClock.instance().now()
|
|
|
|
|
|
def now_ns() -> int:
|
|
"""
|
|
Get current timestamp (nanoseconds since app start).
|
|
|
|
Convenience function for GlobalClock.instance().now_ns()
|
|
"""
|
|
return GlobalClock.instance().now_ns()
|
|
|
|
|
|
# =============================================================================
|
|
# Demo
|
|
# =============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
import time as time_module
|
|
|
|
print("Global Clock Demo")
|
|
print("=" * 50)
|
|
|
|
# Get clock instance
|
|
clock = GlobalClock.instance()
|
|
|
|
print(f"App start time: {clock.now():.6f}s")
|
|
|
|
# Simulate some operations
|
|
time_module.sleep(0.1)
|
|
print(f"After 100ms: {clock.now():.6f}s")
|
|
|
|
time_module.sleep(0.2)
|
|
print(f"After 300ms total: {clock.now():.6f}s")
|
|
|
|
# Nanosecond precision
|
|
print(f"\nNanoseconds: {clock.now_ns()} ns")
|
|
|
|
# Demonstrate synchronization
|
|
print("\n" + "=" * 50)
|
|
print("Synchronization Demo:")
|
|
print("=" * 50)
|
|
|
|
uart_timestamp = clock.now()
|
|
print(f"UART packet at: {uart_timestamp:.6f}s")
|
|
|
|
time_module.sleep(0.05)
|
|
|
|
i2c_timestamp = clock.now()
|
|
print(f"I2C reading at: {i2c_timestamp:.6f}s")
|
|
|
|
print(f"\nTime difference: {(i2c_timestamp - uart_timestamp)*1000:.2f}ms")
|
|
print("✓ Both timestamps use same reference!")
|