161 lines
5.3 KiB
Python
161 lines
5.3 KiB
Python
"""SQLite persistence for waypoints and routes."""
|
|
import sqlite3, json, uuid
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
def _conn(db_path: Path):
|
|
con = sqlite3.connect(db_path)
|
|
con.row_factory = sqlite3.Row
|
|
return con
|
|
|
|
|
|
def init_db(db_path: Path):
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
con = _conn(db_path)
|
|
con.executescript("""
|
|
CREATE TABLE IF NOT EXISTS waypoints (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
lat REAL NOT NULL,
|
|
lon REAL NOT NULL,
|
|
notes TEXT,
|
|
mark_type TEXT DEFAULT '',
|
|
locked INTEGER DEFAULT 0,
|
|
created_at TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS routes (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
wpt_ids TEXT NOT NULL, -- JSON array of waypoint ids in order
|
|
created_at TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS track_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
lat REAL,
|
|
lon REAL,
|
|
sog REAL,
|
|
cog REAL,
|
|
alt REAL,
|
|
hdop REAL,
|
|
ts TEXT
|
|
);
|
|
""")
|
|
con.commit()
|
|
# Migration: add mark_type column to existing DBs that don't have it
|
|
try:
|
|
con.execute("ALTER TABLE waypoints ADD COLUMN mark_type TEXT DEFAULT ''")
|
|
con.commit()
|
|
except Exception:
|
|
pass # column already exists
|
|
try:
|
|
con.execute("ALTER TABLE waypoints ADD COLUMN locked INTEGER DEFAULT 0")
|
|
con.commit()
|
|
except Exception:
|
|
pass
|
|
con.close()
|
|
|
|
|
|
# ── Waypoints ─────────────────────────────────────────────────────────────────
|
|
|
|
def get_waypoints(db_path: Path) -> list:
|
|
con = _conn(db_path)
|
|
rows = con.execute("SELECT * FROM waypoints ORDER BY created_at").fetchall()
|
|
con.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def save_waypoint(db_path: Path, data: dict) -> dict:
|
|
wid = data.get("id") or str(uuid.uuid4())[:8].upper()
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
con = _conn(db_path)
|
|
con.execute("""
|
|
INSERT INTO waypoints (id, name, lat, lon, notes, mark_type, locked, created_at)
|
|
VALUES (?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name=excluded.name, lat=excluded.lat, lon=excluded.lon,
|
|
notes=excluded.notes, mark_type=excluded.mark_type,
|
|
locked=excluded.locked
|
|
""", (wid, data["name"], data["lat"], data["lon"],
|
|
data.get("notes", ""), data.get("mark_type", ""),
|
|
int(data.get("locked", 0)),
|
|
data.get("created_at", now)))
|
|
con.commit()
|
|
con.close()
|
|
return {**data, "id": wid, "created_at": now}
|
|
|
|
|
|
def delete_waypoint(db_path: Path, wid: str):
|
|
con = _conn(db_path)
|
|
con.execute("DELETE FROM waypoints WHERE id=?", (wid,))
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
# ── Routes ────────────────────────────────────────────────────────────────────
|
|
|
|
def get_routes(db_path: Path) -> list:
|
|
con = _conn(db_path)
|
|
rows = con.execute("SELECT * FROM routes ORDER BY created_at").fetchall()
|
|
con.close()
|
|
result = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
d["wpt_ids"] = json.loads(d["wpt_ids"])
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
def save_route(db_path: Path, data: dict) -> dict:
|
|
rid = data.get("id") or str(uuid.uuid4())[:8].upper()
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
con = _conn(db_path)
|
|
con.execute("""
|
|
INSERT INTO routes (id, name, wpt_ids, created_at)
|
|
VALUES (?,?,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name=excluded.name, wpt_ids=excluded.wpt_ids
|
|
""", (rid, data["name"], json.dumps(data.get("wpt_ids", [])),
|
|
data.get("created_at", now)))
|
|
con.commit()
|
|
con.close()
|
|
return {**data, "id": rid, "wpt_ids": data.get("wpt_ids", []), "created_at": now}
|
|
|
|
|
|
def delete_route(db_path: Path, rid: str):
|
|
con = _conn(db_path)
|
|
con.execute("DELETE FROM routes WHERE id=?", (rid,))
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
# ── Track log ─────────────────────────────────────────────────────────────────
|
|
|
|
def log_position(db_path: Path, fix: dict):
|
|
con = _conn(db_path)
|
|
con.execute(
|
|
"INSERT INTO track_log (lat,lon,sog,cog,alt,hdop,ts) VALUES (?,?,?,?,?,?,?)",
|
|
(fix.get("lat"), fix.get("lon"), fix.get("sog"), fix.get("cog"),
|
|
fix.get("altitude"), fix.get("hdop"),
|
|
datetime.now(timezone.utc).isoformat())
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def get_track(db_path: Path, limit: int = 2000) -> list:
|
|
con = _conn(db_path)
|
|
rows = con.execute(
|
|
"SELECT lat,lon,sog,cog,alt,hdop,ts FROM track_log ORDER BY id DESC LIMIT ?",
|
|
(limit,)
|
|
).fetchall()
|
|
con.close()
|
|
return [dict(r) for r in reversed(rows)]
|
|
|
|
|
|
def clear_track(db_path: Path):
|
|
con = _conn(db_path)
|
|
con.execute("DELETE FROM track_log")
|
|
con.commit()
|
|
con.close()
|