57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
 
|
"""Wall-clock helpers for Medulla tick filtering.
|
||
|
|
|
||
|
|
Two tick conventions appear in this dataset:
|
||
|
|
|
||
|
|
- LiDAR DObject ``tic`` / recovered ``indices.log``: ``DateTime.Now.Ticks`` (local)
|
||
|
|
- IMU / MSOP payload host receive fields: UTC ``DateTime.UtcNow.Ticks``
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
|
||
|
|
TICKS_PER_SECOND = 10_000_000
|
||
|
|
DOTNET_UNIX_EPOCH_TICKS = 621355968000000000
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_local_wall(text: str) -> datetime:
|
||
|
|
normalized = text.strip().replace(" ", "T")
|
||
|
|
if normalized.endswith("Z"):
|
||
|
|
raise ValueError("expected local wall time without Z; got UTC marker")
|
||
|
|
if "+" in normalized[10:]:
|
||
|
|
idx = normalized.find("+", 10)
|
||
|
|
normalized = normalized[:idx]
|
||
|
|
elif normalized.count("-") > 2:
|
||
|
|
# timezone like -08:00 after the date
|
||
|
|
idx = normalized.find("-", 10)
|
||
|
|
if idx > 0 and ":" in normalized[idx + 1 :]:
|
||
|
|
normalized = normalized[:idx]
|
||
|
|
return datetime.fromisoformat(normalized).replace(tzinfo=None)
|
||
|
|
|
||
|
|
|
||
|
|
def local_wall_to_dotnet_ticks(text: str) -> int:
|
||
|
|
"""Local wall time → ``DateTime.Now.Ticks`` (LiDAR DObject tic)."""
|
||
|
|
|
||
|
|
dt = _parse_local_wall(text)
|
||
|
|
delta = dt - datetime(1, 1, 1)
|
||
|
|
return int(delta.total_seconds() * TICKS_PER_SECOND)
|
||
|
|
|
||
|
|
|
||
|
|
def local_wall_to_utc_dotnet_ticks(text: str, *, tz_hours: float = 8.0) -> int:
|
||
|
|
"""Local wall time in ``tz_hours`` → UTC ``DateTime.UtcNow.Ticks`` (IMU host)."""
|
||
|
|
|
||
|
|
dt = _parse_local_wall(text).replace(tzinfo=timezone(timedelta(hours=tz_hours)))
|
||
|
|
unix = dt.timestamp()
|
||
|
|
return int(round(unix * TICKS_PER_SECOND)) + DOTNET_UNIX_EPOCH_TICKS
|
||
|
|
|
||
|
|
|
||
|
|
def dotnet_ticks_to_local_iso(ticks: int) -> str:
|
||
|
|
dt = datetime(1, 1, 1) + timedelta(microseconds=ticks / 10.0)
|
||
|
|
return dt.isoformat(timespec="milliseconds")
|
||
|
|
|
||
|
|
|
||
|
|
def utc_dotnet_ticks_to_unix_s(ticks: int) -> float:
|
||
|
|
"""UTC ``DateTime.UtcNow.Ticks`` → Unix seconds."""
|
||
|
|
|
||
|
|
return (float(ticks) - float(DOTNET_UNIX_EPOCH_TICKS)) / float(TICKS_PER_SECOND)
|