chaos/NTP-Timeturner
1
0
Fork 0
mirror of https://github.com/cjfranko/NTP-Timeturner.git synced 2025-11-08 18:32:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update timeturner.py

Browse source
This commit is contained in:
Chris Frankland-Wright 2025-07-07 22:07:01 +01:00 • committed by GitHub
parent 69419b6d59
commit 93d1ac7555
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 17 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

42
timeturner.py
View file

@ -6,15 +6,14 @@ import re
import subprocess import subprocess
import os import os
SERIAL_PORT = None # Will be auto-detected SERIAL_PORT = None
BAUD_RATE = 115200 BAUD_RATE = 115200
FRAME_RATE = 25.0 # Default, will update from stream FRAME_RATE = 25.0
last_ltc_time = None last_ltc_time = None
last_frame = None last_frame = None
lock_count = 0 lock_count = 0
free_count = 0 free_count = 0
sync_requested = False
sync_pending = False sync_pending = False
def find_teensy_serial(): def find_teensy_serial():
@ -24,10 +23,6 @@ def find_teensy_serial():
return None return None
def parse_ltc_line(line): def parse_ltc_line(line):
"""
Expected format: [LOCK] 19:56:56:14 | 25.00fps
or dropframe style: [LOCK] 19:56:56;14 | 29.97fps
"""
match = re.match(r"\[(LOCK|FREE)\]\s+(\d{2}):(\d{2}):(\d{2})[:;](\d{2})\s+\|\s+([\d.]+)fps", line) match = re.match(r"\[(LOCK|FREE)\]\s+(\d{2}):(\d{2}):(\d{2})[:;](\d{2})\s+\|\s+([\d.]+)fps", line)
if not match: if not match:
return None return None
@ -45,19 +40,18 @@ def get_system_time():
return datetime.datetime.now() return datetime.datetime.now()
def format_time(dt): def format_time(dt):
return dt.strftime("%H:%M:%S.%f")[:-3] # show milliseconds only return dt.strftime("%H:%M:%S.%f")[:-3]
def run_curses(stdscr): def run_curses(stdscr):
global last_ltc_time, FRAME_RATE, lock_count, free_count, sync_requested, sync_pending, SERIAL_PORT global last_ltc_time, FRAME_RATE, lock_count, free_count, sync_pending, SERIAL_PORT
curses.curs_set(0) curses.curs_set(0)
stdscr.nodelay(True) stdscr.nodelay(True)
stdscr.timeout(100) stdscr.timeout(100)
# Auto-detect Teensy serial
SERIAL_PORT = find_teensy_serial() SERIAL_PORT = find_teensy_serial()
if not SERIAL_PORT: if not SERIAL_PORT:
stdscr.addstr(0, 0, "❌ No serial device found. Connect Teensy and try again.") stdscr.addstr(0, 0, "No serial device found.")
stdscr.refresh() stdscr.refresh()
time.sleep(3) time.sleep(3)
return return
@ -65,27 +59,27 @@ def run_curses(stdscr):
try: try:
ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1) ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1)
except Exception as e: except Exception as e:
stdscr.addstr(0, 0, f"❌ Failed to open serial port: {e}") stdscr.addstr(0, 0, f"Error opening serial: {e}")
stdscr.refresh() stdscr.refresh()
time.sleep(3) time.sleep(3)
return return
while True: while True:
try: try:
pre_read_time = datetime.datetime.now()
line = ser.readline().decode(errors='ignore').strip() line = ser.readline().decode(errors='ignore').strip()
parsed = parse_ltc_line(line) parsed = parse_ltc_line(line)
if parsed: if parsed:
FRAME_RATE = parsed["frame_rate"]
if parsed["status"] == "LOCK": if parsed["status"] == "LOCK":
lock_count += 1 lock_count += 1
else: else:
free_count += 1 free_count += 1
FRAME_RATE = parsed["frame_rate"]
# Construct full datetime with microsecond precision from LTC
ms = int((parsed["frames"] / FRAME_RATE) * 1000) ms = int((parsed["frames"] / FRAME_RATE) * 1000)
ltc_dt = datetime.datetime.now().replace( ltc_dt = pre_read_time.replace(
hour=parsed["hours"], hour=parsed["hours"],
minute=parsed["minutes"], minute=parsed["minutes"],
second=parsed["seconds"], second=parsed["seconds"],
@ -95,29 +89,27 @@ def run_curses(stdscr):
last_ltc_time = ltc_dt last_ltc_time = ltc_dt
last_frame = parsed["frames"] last_frame = parsed["frames"]
# Perform sync only if requested and we just got a new frame
if sync_pending: if sync_pending:
do_sync(stdscr, ltc_dt) do_sync(stdscr, ltc_dt)
sync_pending = False sync_pending = False
# Display # Drawing UI
stdscr.erase() stdscr.erase()
stdscr.addstr(0, 2, f"NTP Timeturner v0.7") stdscr.addstr(0, 2, f"NTP Timeturner v0.8")
stdscr.addstr(1, 2, f"Using Serial Port: {SERIAL_PORT}") stdscr.addstr(1, 2, f"Using Serial Port: {SERIAL_PORT}")
stdscr.addstr(3, 2, f"LTC Status : {parsed['status'] if parsed else '…'}") stdscr.addstr(3, 2, f"LTC Status : {parsed['status'] if parsed else '…'}")
stdscr.addstr(4, 2, f"LTC Timecode : {parsed['hours']:02}:{parsed['minutes']:02}:{parsed['seconds']:02}:{parsed['frames']:02}" if parsed else "LTC Timecode : …") stdscr.addstr(4, 2, f"LTC Timecode : {parsed['hours']:02}:{parsed['minutes']:02}:{parsed['seconds']:02}:{parsed['frames']:02}" if parsed else "LTC Timecode : …")
stdscr.addstr(5, 2, f"Frame Rate : {FRAME_RATE:.2f}fps") stdscr.addstr(5, 2, f"Frame Rate : {FRAME_RATE:.2f}fps")
now = get_system_time() now = get_system_time()
stdscr.addstr(6, 2, f"System Clock : {format_time(now)}") stdscr.addstr(6, 2, f"System Clock : {format_time(now)}")
# Sync offset
if last_ltc_time: if last_ltc_time:
offset_ms = (now - last_ltc_time).total_seconds() * 1000 offset_ms = (now - last_ltc_time).total_seconds() * 1000
offset_frames = offset_ms / (1000 / FRAME_RATE) offset_frames = offset_ms / (1000 / FRAME_RATE)
stdscr.addstr(7, 2, f"Sync Offset : {offset_ms:+.0f} ms ({offset_frames:+.0f} frames)") stdscr.addstr(7, 2, f"Sync Offset : {offset_ms:+.0f} ms ({offset_frames:+.0f} frames)")
stdscr.addstr(8, 2, f"Lock Ratio : {lock_count} LOCK / {free_count} FREE") stdscr.addstr(8, 2, f"Lock Ratio : {lock_count} LOCK / {free_count} FREE")
stdscr.addstr(10, 2, "[S] Set system clock to LTC [Ctrl+C] Quit") stdscr.addstr(10, 2, "[S] Set system clock to LTC [Ctrl+C] Quit")
stdscr.refresh() stdscr.refresh()
@ -128,17 +120,17 @@ def run_curses(stdscr):
except KeyboardInterrupt: except KeyboardInterrupt:
break break
except Exception as e: except Exception as e:
stdscr.addstr(15, 2, f"⚠️ Error: {e}") stdscr.addstr(13, 2, f"⚠️ Error: {e}")
stdscr.refresh() stdscr.refresh()
time.sleep(1) time.sleep(1)
def do_sync(stdscr, ltc_dt): def do_sync(stdscr, ltc_dt):
try: try:
timestamp = ltc_dt.strftime("%H:%M:%S.%f")[:-3] # Drop microsecond to milliseconds timestamp = ltc_dt.strftime("%H:%M:%S.%f")[:-3]
subprocess.run(["sudo", "date", "-s", timestamp], check=True) subprocess.run(["sudo", "date", "-s", timestamp], check=True)
stdscr.addstr(13, 2, f"✔️ System clock set to LTC: {timestamp}") stdscr.addstr(13, 2, f"✔️ System clock set to: {timestamp}")
except Exception as e: except Exception as e:
stdscr.addstr(13, 2, f"❌ Failed to sync system time: {e}") stdscr.addstr(13, 2, f"❌ Sync failed: {e}")
if __name__ == "__main__": if __name__ == "__main__":
curses.wrapper(run_curses) curses.wrapper(run_curses)