Add sensor freshness checks
This commit is contained in:
1 parent
c434f3995d
commit
cea79b6b48
12 files changed
+213
-25
No files matched your search
@@ -0,0 +1,18 @@
|
||||
"""Pure timestamp validation for reported sensor snapshots."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from math import isfinite
|
||||
|
||||
|
||||
def freshness(reported: datetime | None, now: datetime, max_age_seconds: float) -> dict:
|
||||
"""Use report time, not value-change time; reject unknown/future clocks."""
|
||||
if not isfinite(max_age_seconds) or max_age_seconds <= 0:
|
||||
raise ValueError("Freshness limit must be positive and finite")
|
||||
if not isinstance(reported, datetime) or reported.tzinfo is None:
|
||||
return {"status": "invalid_timestamp", "age_seconds": None}
|
||||
age = (now - reported).total_seconds()
|
||||
if age < 0:
|
||||
return {"status": "future_timestamp", "age_seconds": round(age, 1)}
|
||||
return {"status": "stale" if age > max_age_seconds else "valid",
|
||||
"age_seconds": round(age, 1)}
|
||||
Reference in new issue
Block a user