118 lines
3.5 KiB
Python
Executable File
118 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
# Export traccar tc_positions table as CSV (e.g. using Adminer)
|
|
FILE="traccar-export-example.csv"
|
|
|
|
|
|
import csv
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
# traccar_id --> topic
|
|
tcid_map = {
|
|
1: 'owntracks/mb/iPhone 13 Pro',
|
|
2: 'owntracks/mb/iPhone XS',
|
|
3: 'owntracks/mb/iPhone 6S',
|
|
4: 'Google Latitude',
|
|
5: 'owntracks/mb/iPhone 16 Pro'
|
|
}
|
|
|
|
with open(FILE, "rt", encoding="utf-8-sig") as f:
|
|
csvreader = csv.DictReader(f)
|
|
i = 0
|
|
for row in csvreader:
|
|
attrs = json.loads(row["attributes"])
|
|
|
|
# OwnTracks format: https://owntracks.org/booklet/tech/json/#_typelocation
|
|
owntr = {
|
|
"_type": "location",
|
|
"acc": int(row["accuracy"]),
|
|
"alt": int(row["altitude"]),
|
|
"bs": 1, # 0=unknown, 1=unplugged, 2=charging, 3=full
|
|
"lat": float(row["latitude"]),
|
|
"lon": float(row["longitude"]),
|
|
"t": "p", # p=ping, u=manual, t=timer, etc.
|
|
"tid": "MB",
|
|
"tst": int(datetime.strptime(row["fixtime"], "%Y-%m-%d %H:%M:%S").timestamp()),
|
|
"topic": tcid_map[int(row["deviceid"])],
|
|
}
|
|
if "batteryLevel" in attrs:
|
|
owntr["batt"] = int(attrs["batteryLevel"])
|
|
|
|
if float(row["speed"]) > 0:
|
|
owntr["vel"] = float(row["speed"])
|
|
owntr["cog"] = int(row["course"])
|
|
|
|
#print("### Source data:")
|
|
#print(repr(row))
|
|
|
|
#print("### Output:")
|
|
#print(json.dumps(owntr))
|
|
|
|
dwirec = {
|
|
"battery_status": 1,
|
|
"ping": None,
|
|
"battery": owntr["batt"] if "batt" in owntr else None,
|
|
"tracker_id": owntr["tid"],
|
|
"topic": owntr["topic"],
|
|
"altitude": owntr["alt"],
|
|
"longitude": owntr["lon"],
|
|
"velocity": str(owntr["vel"]) if "vel" in owntr else None,
|
|
"trigger": 0,
|
|
"bssid": None,
|
|
"ssid": None,
|
|
"connection": 0,
|
|
"vertical_accuracy": None,
|
|
"accuracy": owntr["acc"],
|
|
"timestamp": owntr["tst"],
|
|
"latitude": owntr["lat"],
|
|
"mode": None,
|
|
"inrids": None,
|
|
"in_regions": None,
|
|
"raw_data": json.dumps(owntr),
|
|
"import_id": None,
|
|
"city": None,
|
|
"country" :None,
|
|
"created_at": row["fixtime"],
|
|
"updated_at": row["fixtime"],
|
|
"user_id": 1,
|
|
"geodata": "{}",
|
|
"visit_id": None,
|
|
"reverse_geocoded_at": None,
|
|
"course": owntr["cog"] if "cog" in owntr else None,
|
|
"course_accuracy": None,
|
|
"external_track_id": None,
|
|
}
|
|
|
|
if i%10 == 0:
|
|
if i>0:
|
|
print(";")
|
|
keys = dwirec.keys()
|
|
keystring = '"' + '", "'.join(keys) + '"'
|
|
print(f'INSERT INTO "points" ({keystring}) VALUES ')
|
|
elif i>0:
|
|
print(",")
|
|
|
|
values = []
|
|
for k in keys:
|
|
v = dwirec[k]
|
|
if type(v) is str:
|
|
if "'" in v:
|
|
v = v.replace("'", "\\'")
|
|
values.append("'" + v + "'")
|
|
elif v is None:
|
|
values.append("NULL")
|
|
else:
|
|
values.append(str(v))
|
|
|
|
valuestring = ", ".join(values)
|
|
print(f"({valuestring})", end="")
|
|
|
|
i += 1
|
|
#if i > 10:
|
|
# break
|
|
|
|
print(";")
|
|
print(f"{f} records converted.", file=sys.stderr)
|