Added telegraf configs
This commit is contained in:
333
telegraf/gpsdMeasurement.py
Normal file
333
telegraf/gpsdMeasurement.py
Normal file
@@ -0,0 +1,333 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import math
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
# offset_ns is computed as:
|
||||
|
||||
# (clock_sec * 1e9 + clock_nsec) - (real_sec * 1e9 + real_nsec)
|
||||
|
||||
|
||||
GPSPIPE_CMD = ["gpspipe", "-w", "--json"]
|
||||
HOST_TAG = "ntprana"
|
||||
|
||||
|
||||
running = True
|
||||
|
||||
|
||||
def handle_signal(signum, frame):
|
||||
global running
|
||||
running = False
|
||||
|
||||
|
||||
signal.signal(signal.SIGINT, handle_signal)
|
||||
signal.signal(signal.SIGTERM, handle_signal)
|
||||
|
||||
|
||||
def esc_tag(value: Any) -> str:
|
||||
s = str(value)
|
||||
return s.replace("\\", "\\\\").replace(" ", r"\ ").replace(",", r"\,").replace("=", r"\=")
|
||||
|
||||
|
||||
def esc_field_str(value: Any) -> str:
|
||||
s = str(value)
|
||||
s = s.replace("\\", "\\\\").replace('"', r"\"")
|
||||
return f'"{s}"'
|
||||
|
||||
|
||||
def is_num(value: Any) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
|
||||
|
||||
|
||||
def bool_to_field(value: bool) -> str:
|
||||
return "true" if value else "false"
|
||||
|
||||
|
||||
def build_line(measurement: str, tags: Dict[str, Any], fields: Dict[str, Any], ts_ns: Optional[int] = None) -> Optional[str]:
|
||||
clean_fields = []
|
||||
for key, value in fields.items():
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, bool):
|
||||
clean_fields.append(f"{key}={bool_to_field(value)}")
|
||||
elif isinstance(value, int) and not isinstance(value, bool):
|
||||
clean_fields.append(f"{key}={value}i")
|
||||
elif is_num(value):
|
||||
clean_fields.append(f"{key}={value}")
|
||||
else:
|
||||
clean_fields.append(f"{key}={esc_field_str(value)}")
|
||||
|
||||
if not clean_fields:
|
||||
return None
|
||||
|
||||
tag_str = ",".join(f"{esc_tag(k)}={esc_tag(v)}" for k, v in tags.items() if v is not None)
|
||||
line = measurement
|
||||
if tag_str:
|
||||
line += "," + tag_str
|
||||
line += " " + ",".join(clean_fields)
|
||||
if ts_ns is not None:
|
||||
line += f" {ts_ns}"
|
||||
return line
|
||||
|
||||
|
||||
def gps_constellation_name(gnssid: Optional[int]) -> str:
|
||||
mapping = {
|
||||
0: "gps",
|
||||
1: "sbas",
|
||||
2: "galileo",
|
||||
3: "beidou",
|
||||
4: "imes",
|
||||
5: "qzss",
|
||||
6: "glonass",
|
||||
}
|
||||
return mapping.get(gnssid, "unknown")
|
||||
|
||||
|
||||
def summarize_satellites(satellites: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
out: Dict[str, Any] = {
|
||||
"gps_visible": 0,
|
||||
"gps_used": 0,
|
||||
"sbas_visible": 0,
|
||||
"sbas_used": 0,
|
||||
"galileo_visible": 0,
|
||||
"galileo_used": 0,
|
||||
"beidou_visible": 0,
|
||||
"beidou_used": 0,
|
||||
"qzss_visible": 0,
|
||||
"qzss_used": 0,
|
||||
"glonass_visible": 0,
|
||||
"glonass_used": 0,
|
||||
"unknown_visible": 0,
|
||||
"unknown_used": 0,
|
||||
"used_signal_sum": 0.0,
|
||||
"used_signal_count": 0,
|
||||
"visible_signal_sum": 0.0,
|
||||
"visible_signal_count": 0,
|
||||
"max_signal": 0.0,
|
||||
}
|
||||
|
||||
for sat in satellites:
|
||||
constellation = gps_constellation_name(sat.get("gnssid"))
|
||||
ss = sat.get("ss", 0.0)
|
||||
used = bool(sat.get("used", False))
|
||||
|
||||
vis_key = f"{constellation}_visible"
|
||||
used_key = f"{constellation}_used"
|
||||
|
||||
if vis_key not in out:
|
||||
vis_key = "unknown_visible"
|
||||
used_key = "unknown_used"
|
||||
|
||||
out[vis_key] += 1
|
||||
if used:
|
||||
out[used_key] += 1
|
||||
|
||||
if is_num(ss):
|
||||
out["visible_signal_sum"] += float(ss)
|
||||
out["visible_signal_count"] += 1
|
||||
out["max_signal"] = max(out["max_signal"], float(ss))
|
||||
if used:
|
||||
out["used_signal_sum"] += float(ss)
|
||||
out["used_signal_count"] += 1
|
||||
|
||||
out["avg_signal_visible"] = (
|
||||
out["visible_signal_sum"] / out["visible_signal_count"] if out["visible_signal_count"] else None
|
||||
)
|
||||
out["avg_signal_used"] = (
|
||||
out["used_signal_sum"] / out["used_signal_count"] if out["used_signal_count"] else None
|
||||
)
|
||||
|
||||
# cleanup helper sums/counters from final output
|
||||
del out["visible_signal_sum"]
|
||||
del out["visible_signal_count"]
|
||||
del out["used_signal_sum"]
|
||||
del out["used_signal_count"]
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def pps_offset_ns(pps: Dict[str, Any]) -> Optional[int]:
|
||||
real_sec = pps.get("real_sec")
|
||||
real_nsec = pps.get("real_nsec")
|
||||
clock_sec = pps.get("clock_sec")
|
||||
clock_nsec = pps.get("clock_nsec")
|
||||
|
||||
if not all(isinstance(v, int) for v in [real_sec, real_nsec, clock_sec, clock_nsec]):
|
||||
return None
|
||||
|
||||
real_total = real_sec * 1_000_000_000 + real_nsec
|
||||
clock_total = clock_sec * 1_000_000_000 + clock_nsec
|
||||
return clock_total - real_total
|
||||
|
||||
|
||||
def main():
|
||||
global running
|
||||
|
||||
proc = subprocess.Popen(
|
||||
GPSPIPE_CMD,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
)
|
||||
|
||||
latest_tpv: Optional[Dict[str, Any]] = None
|
||||
latest_sky: Optional[Dict[str, Any]] = None
|
||||
latest_pps: Optional[Dict[str, Any]] = None
|
||||
last_emit_sec: Optional[int] = None
|
||||
# NGL at some point I just put cloudee to do it bc I've wrote simmilar code before many times
|
||||
try:
|
||||
assert proc.stdout is not None
|
||||
while running:
|
||||
line = proc.stdout.readline()
|
||||
if not line:
|
||||
if proc.poll() is not None:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
continue
|
||||
|
||||
line = line.strip()
|
||||
if not line or not line.startswith("{"):
|
||||
continue
|
||||
|
||||
try:
|
||||
msg = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
cls = msg.get("class")
|
||||
|
||||
if cls == "TPV":
|
||||
latest_tpv = msg
|
||||
elif cls == "SKY":
|
||||
if latest_sky is None:
|
||||
latest_sky = msg
|
||||
else:
|
||||
merged = latest_sky.copy()
|
||||
merged.update(msg)
|
||||
|
||||
# Keep the richer satellites list if the new SKY message doesn't include it
|
||||
if "satellites" not in msg and "satellites" in latest_sky:
|
||||
merged["satellites"] = latest_sky["satellites"]
|
||||
|
||||
# Keep nSat/uSat if the new SKY message doesn't include them
|
||||
if "nSat" not in msg and "nSat" in latest_sky:
|
||||
merged["nSat"] = latest_sky["nSat"]
|
||||
|
||||
if "uSat" not in msg and "uSat" in latest_sky:
|
||||
merged["uSat"] = latest_sky["uSat"]
|
||||
|
||||
latest_sky = merged
|
||||
elif cls == "PPS":
|
||||
latest_pps = msg
|
||||
else:
|
||||
continue
|
||||
|
||||
now_sec = int(time.time())
|
||||
if last_emit_sec == now_sec:
|
||||
continue
|
||||
|
||||
lines_out = []
|
||||
ts_ns = time.time_ns()
|
||||
|
||||
common_tags = {
|
||||
"host": HOST_TAG,
|
||||
"device": (latest_tpv or latest_sky or latest_pps or {}).get("device", "unknown"),
|
||||
}
|
||||
|
||||
if latest_tpv:
|
||||
fix_fields = {
|
||||
"mode": latest_tpv.get("mode"),
|
||||
"lat": latest_tpv.get("lat"),
|
||||
"lon": latest_tpv.get("lon"),
|
||||
"alt": latest_tpv.get("alt"),
|
||||
"alt_msl": latest_tpv.get("altMSL"),
|
||||
"alt_hae": latest_tpv.get("altHAE"),
|
||||
"speed": latest_tpv.get("speed"),
|
||||
"climb": latest_tpv.get("climb"),
|
||||
"track": latest_tpv.get("track"),
|
||||
"magtrack": latest_tpv.get("magtrack"),
|
||||
"magvar": latest_tpv.get("magvar"),
|
||||
"ept": latest_tpv.get("ept"),
|
||||
"epx": latest_tpv.get("epx"),
|
||||
"epy": latest_tpv.get("epy"),
|
||||
"epv": latest_tpv.get("epv"),
|
||||
"eph": latest_tpv.get("eph"),
|
||||
"sep": latest_tpv.get("sep"),
|
||||
"ecefpAcc": latest_tpv.get("ecefpAcc"),
|
||||
"ecefvAcc": latest_tpv.get("ecefvAcc"),
|
||||
"leapseconds": latest_tpv.get("leapseconds"),
|
||||
"fix_valid": latest_tpv.get("mode", 0) >= 2,
|
||||
"fix_3d": latest_tpv.get("mode", 0) == 3,
|
||||
}
|
||||
line_fix = build_line("gps_fix", common_tags, fix_fields, ts_ns)
|
||||
if line_fix:
|
||||
lines_out.append(line_fix)
|
||||
|
||||
if latest_sky:
|
||||
satellites = latest_sky.get("satellites", [])
|
||||
sat_summary = summarize_satellites(satellites) if isinstance(satellites, list) else {}
|
||||
|
||||
n_sat = latest_sky.get("nSat")
|
||||
u_sat = latest_sky.get("uSat")
|
||||
sat_used_ratio = None
|
||||
if isinstance(n_sat, int) and n_sat > 0 and isinstance(u_sat, int):
|
||||
sat_used_ratio = u_sat / n_sat
|
||||
|
||||
sky_fields = {
|
||||
"nSat": n_sat,
|
||||
"uSat": u_sat,
|
||||
"sat_used_ratio": sat_used_ratio,
|
||||
"gdop": latest_sky.get("gdop"),
|
||||
"hdop": latest_sky.get("hdop"),
|
||||
"pdop": latest_sky.get("pdop"),
|
||||
"tdop": latest_sky.get("tdop"),
|
||||
"vdop": latest_sky.get("vdop"),
|
||||
"xdop": latest_sky.get("xdop"),
|
||||
"ydop": latest_sky.get("ydop"),
|
||||
**sat_summary,
|
||||
}
|
||||
line_sky = build_line("gps_sky", common_tags, sky_fields, ts_ns)
|
||||
if line_sky:
|
||||
lines_out.append(line_sky)
|
||||
|
||||
if latest_pps:
|
||||
offset_ns = pps_offset_ns(latest_pps)
|
||||
pps_fields = {
|
||||
"offset_ns": offset_ns,
|
||||
"precision_exp": latest_pps.get("precision"),
|
||||
"present": True,
|
||||
}
|
||||
line_pps = build_line(
|
||||
"gps_pps",
|
||||
{
|
||||
"host": HOST_TAG,
|
||||
"device": latest_pps.get("device", "/dev/pps0"),
|
||||
},
|
||||
pps_fields,
|
||||
ts_ns,
|
||||
)
|
||||
if line_pps:
|
||||
lines_out.append(line_pps)
|
||||
|
||||
if lines_out:
|
||||
sys.stdout.write("\n".join(lines_out) + "\n")
|
||||
sys.stdout.flush()
|
||||
last_emit_sec = now_sec
|
||||
|
||||
finally:
|
||||
if proc.poll() is None:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user