export_telemetry_csv.py

#!/usr/bin/env python3
import argparse
import csv
import sqlite3
import sys
from pathlib import Path

script_dir = Path(__file__).parent.resolve()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export telemetry rows from groundstation.db to a CSV file."
    )
    parser.add_argument(
        "--db",
        default=str(script_dir / Path("data") / "groundstation.db"),
        help="Path to the SQLite DB (default: data/groundstation.db)",
    )
    parser.add_argument(
        "--out",
        default=str(script_dir / "telemetry.csv"),
        help="Output CSV path (default: telemetry.csv)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    db_path = Path(args.db)
    if not db_path.exists():
        raise SystemExit(f"DB not found: {db_path}")

    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    with sqlite3.connect(str(db_path)) as conn:
        conn.row_factory = sqlite3.Row
        table_cols = {
            row["name"]
            for row in conn.execute("PRAGMA table_info(telemetry)").fetchall()
        }
        select_cols = [
            "timestamp_ms",
            "data_type",
            "sender_id" if "sender_id" in table_cols else "NULL AS sender_id",
            "values_json" if "values_json" in table_cols else "NULL AS values_json",
            "payload_json" if "payload_json" in table_cols else "NULL AS payload_json",
        ]
        query = (
            "SELECT "
            + ", ".join(select_cols)
            + " FROM telemetry ORDER BY timestamp_ms"
        )
        cursor = conn.execute(query)
        col_names = [col[0] for col in cursor.description]

        with out_path.open("w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(col_names)
            for row in cursor:
                writer.writerow([row[k] for k in col_names])

    print(f"Wrote telemetry CSV to {out_path}")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nExport interrupted.", file=sys.stderr)
        raise SystemExit(130)
    except sqlite3.OperationalError as e:
        print(f"Error: SQLite operation failed: {e}", file=sys.stderr)
        print("Hint: ensure the DB file exists and is not locked by another process.", file=sys.stderr)
        raise SystemExit(1)
    except PermissionError as e:
        print(f"Error: Permission denied: {e}", file=sys.stderr)
        print("Hint: verify read access to DB path and write access to output directory.", file=sys.stderr)
        raise SystemExit(1)
    except FileNotFoundError as e:
        missing = e.filename or "<unknown>"
        print(f"Error: Missing file: {missing}", file=sys.stderr)
        raise SystemExit(1)
    except Exception as e:
        print(f"Error: export_telemetry_csv failed unexpectedly: {e}", file=sys.stderr)
        raise SystemExit(1)