merge

gpx2owntracks.py

@@ -0,0 +1,500 @@
+#!/usr/bin/env python3
+"""
+GPX → OwnTracks (Recorder) importer
+
+Converts .gpx tracks (trk/seg/pt) to OwnTracks Location JSON messages and either:
+  • writes a JSONL file (one message per line), or
+  • writes individual .json files in a Recorder-like folder structure, or
+  • POSTs messages to an OwnTracks Recorder HTTP endpoint (/pub), optionally throttled.
+
+Features
+- Config via YAML + CLI flags that override config
+- Robust error handling & validation
+- Optional speed calculation (m/s) and altitude inclusion
+- Time window filtering
+- Dry-run mode
+- Logging with levels
+
+OwnTracks message fields used
+  _type: "location"
+  tst:   Unix timestamp (UTC)
+  lat, lon, alt (optional), acc (accuracy meters), vel (optional m/s), tid (2-char), conn ("w")
+
+Dependencies
+  pip install gpxpy pyyaml requests
+
+Usage examples
+  # Basic: create JSONL from GPX
+  python gpx2owntracks.py input.gpx --jsonl-path out/track.jsonl --user alice --device pixel --tid AL
+
+  # Using YAML config (CLI overrides YAML)
+  python gpx2owntracks.py input.gpx --config config.yaml --mode http --rate 5
+
+  # Write Recorder-like file tree
+  python gpx2owntracks.py input.gpx --mode files --files-base-dir /var/lib/ot-recorder/store/recorder \
+      --user alice --device pixel
+
+  # Post directly to Recorder (HTTP /pub)
+  python gpx2owntracks.py input.gpx --mode http --http-url https://rec.example.com/pub \
+      --http-user user --http-pass pass --no-verify
+
+Example config.yaml
+-------------------
+user: alice
+device: pixel
+Tid: AL  # two characters (case-sensitive)
+default_acc: 10
+include_alt: true
+calc_speed: true
+assume_timezone: UTC  # used only if GPX points lack timezone info
+mode: jsonl          # jsonl | files | http
+jsonl_path: ./out/track.jsonl
+files_base_dir: /var/lib/ot-recorder/store/recorder
+http:
+  url: https://rec.example.com/pub
+  username: user
+  password: pass
+  tls_verify: true
+rate_limit_per_sec: 5     # for http mode
+start_time: null          # ISO8601 filter (inclusive), e.g. "2025-08-01T00:00:00Z"
+end_time: null            # ISO8601 filter (exclusive)
+log_level: INFO
+
+Recorder file mode layout
+  <files_base_dir>/<user>/<device>/<YYYY>/<MM>/<DD>/<tst>.json
+
+Note: When writing files, Recorder will pick them up if they are placed under its store path.
+
+"""
+from __future__ import annotations
+
+import argparse
+import dataclasses
+import datetime as dt
+import json
+import logging
+import math
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Iterable, List, Optional, Tuple
+
+import gpxpy
+import gpxpy.gpx
+import yaml
+
+# Optional imports for HTTP mode
+try:
+    import requests  # type: ignore
+except Exception:  # pragma: no cover
+    requests = None
+
+ISO8601 = "%Y-%m-%dT%H:%M:%S%z"
+
+
+@dataclasses.dataclass
+class Config:
+    user: Optional[str] = None
+    device: Optional[str] = None
+    tid: Optional[str] = None
+    default_acc: float = 10.0
+    include_alt: bool = True
+    calc_speed: bool = False
+    assume_timezone: str = "UTC"
+    mode: str = "jsonl"  # jsonl | files | http
+    jsonl_path: Optional[str] = None
+    files_base_dir: Optional[str] = None
+    http_url: Optional[str] = None
+    http_username: Optional[str] = None
+    http_password: Optional[str] = None
+    http_tls_verify: bool = True
+    rate_limit_per_sec: Optional[float] = None
+    start_time: Optional[str] = None  # ISO8601
+    end_time: Optional[str] = None    # ISO8601
+    log_level: str = "INFO"
+
+
+class ConfigError(Exception):
+    pass
+
+
+def load_config(path: Optional[str]) -> Config:
+    if not path:
+        return Config()
+    p = Path(path)
+    if not p.exists():
+        raise ConfigError(f"Config file not found: {p}")
+    try:
+        with p.open("r", encoding="utf-8") as f:
+            data = yaml.safe_load(f) or {}
+    except Exception as e:
+        raise ConfigError(f"Failed to parse YAML: {e}")
+
+    def get_bool(d, k, default):
+        v = d.get(k, default)
+        return bool(v)
+
+    http = data.get("http", {}) or {}
+
+    cfg = Config(
+        user=data.get("user"),
+        device=data.get("device"),
+        tid=data.get("Tid") or data.get("tid"),
+        default_acc=float(data.get("default_acc", 10.0)),
+        include_alt=get_bool(data, "include_alt", True),
+        calc_speed=get_bool(data, "calc_speed", False),
+        assume_timezone=str(data.get("assume_timezone", "UTC")),
+        mode=str(data.get("mode", "jsonl")),
+        jsonl_path=data.get("jsonl_path"),
+        files_base_dir=data.get("files_base_dir"),
+        http_url=http.get("url") or data.get("http_url"),
+        http_username=http.get("username") or data.get("http_username"),
+        http_password=http.get("password") or data.get("http_password"),
+        http_tls_verify=bool(http.get("tls_verify", data.get("http_tls_verify", True))),
+        rate_limit_per_sec=(float(data.get("rate_limit_per_sec")) if data.get("rate_limit_per_sec") is not None else None),
+        start_time=data.get("start_time"),
+        end_time=data.get("end_time"),
+        log_level=str(data.get("log_level", "INFO")).upper(),
+    )
+    return cfg
+
+
+def merge_cli_over_config(cfg: Config, args: argparse.Namespace) -> Config:
+    # Helper to prefer CLI when provided
+    def override(val, cli_val):
+        return cli_val if cli_val is not None else val
+
+    cfg.user = override(cfg.user, args.user)
+    cfg.device = override(cfg.device, args.device)
+    cfg.tid = override(cfg.tid, args.tid)
+    cfg.default_acc = override(cfg.default_acc, args.acc)
+    if args.include_alt is not None:
+        cfg.include_alt = args.include_alt
+    if args.calc_speed is not None:
+        cfg.calc_speed = args.calc_speed
+    cfg.assume_timezone = override(cfg.assume_timezone, args.assume_timezone)
+    cfg.mode = override(cfg.mode, args.mode)
+    cfg.jsonl_path = override(cfg.jsonl_path, args.jsonl_path)
+    cfg.files_base_dir = override(cfg.files_base_dir, args.files_base_dir)
+    cfg.http_url = override(cfg.http_url, args.http_url)
+    cfg.http_username = override(cfg.http_username, args.http_user)
+    cfg.http_password = override(cfg.http_password, args.http_pass)
+    if args.verify is not None:
+        cfg.http_tls_verify = args.verify
+    cfg.rate_limit_per_sec = override(cfg.rate_limit_per_sec, args.rate)
+    cfg.start_time = override(cfg.start_time, args.start_time)
+    cfg.end_time = override(cfg.end_time, args.end_time)
+    cfg.log_level = override(cfg.log_level, args.log_level.upper() if args.log_level else None)
+    return cfg
+
+
+class OwnTracksMessage(dict):
+    """Typed helper for OwnTracks location message."""
+
+    @staticmethod
+    def from_point(point: gpxpy.gpx.GPXTrackPoint, *,
+                   default_acc: float = 10.0,
+                   include_alt: bool = True,
+                   vel: Optional[float] = None,
+                   tid: Optional[str] = None) -> "OwnTracksMessage":
+        if not point.time:
+            raise ValueError("GPX point missing timestamp")
+        # Ensure timezone-aware UTC
+        if point.time.tzinfo is None:
+            # assume input is UTC naive
+            t_utc = point.time.replace(tzinfo=dt.timezone.utc)
+        else:
+            t_utc = point.time.astimezone(dt.timezone.utc)
+        tst = int(t_utc.timestamp())
+
+        msg: OwnTracksMessage = OwnTracksMessage({
+            "_type": "location",
+            "tst": tst,
+            "lat": float(point.latitude),
+            "lon": float(point.longitude),
+            "acc": float(default_acc),
+            "conn": "w",
+        })
+        if include_alt and point.elevation is not None and not math.isnan(point.elevation):
+            msg["alt"] = float(point.elevation)
+        if vel is not None and not math.isnan(vel):
+            msg["vel"] = float(max(0.0, vel))
+        if tid:
+            if len(tid) != 2:
+                raise ValueError("tid must be exactly 2 characters, e.g. 'AL'")
+            msg["tid"] = tid
+        return msg
+
+
+@dataclasses.dataclass
+class OutputSink:
+    mode: str
+    jsonl_path: Optional[Path] = None
+    base_dir: Optional[Path] = None
+    http_url: Optional[str] = None
+    http_auth: Optional[Tuple[str, str]] = None
+    http_verify: bool = True
+    rate_limit_per_sec: Optional[float] = None
+    user: Optional[str] = None
+    device: Optional[str] = None
+    dry_run: bool = False
+
+    def open(self):
+        if self.mode == "jsonl" and self.jsonl_path and not self.dry_run:
+            self.jsonl_path.parent.mkdir(parents=True, exist_ok=True)
+            self._jsonl_file = self.jsonl_path.open("w", encoding="utf-8")
+        else:
+            self._jsonl_file = None
+
+        if self.mode == "http":
+            if requests is None:
+                raise ConfigError("requests not available; install it or use jsonl/files mode")
+            if not self.http_url:
+                raise ConfigError("HTTP mode requires http_url")
+
+    def close(self):
+        if hasattr(self, "_jsonl_file") and self._jsonl_file:
+            self._jsonl_file.close()
+
+    def _rate_limit(self):
+        if self.rate_limit_per_sec and self.rate_limit_per_sec > 0:
+            time.sleep(1.0 / self.rate_limit_per_sec)
+
+    def write(self, msg: OwnTracksMessage):
+        if self.mode == "jsonl":
+            line = json.dumps(msg, separators=(",", ":"))
+            if self.dry_run:
+                logging.debug(line)
+            else:
+                assert self._jsonl_file is not None
+                self._jsonl_file.write(line + "\n")
+
+        elif self.mode == "files":
+            if not (self.base_dir and self.user and self.device):
+                raise ConfigError("files mode requires files_base_dir, user, and device")
+            if "tst" not in msg:
+                raise ValueError("message missing tst")
+            t = dt.datetime.fromtimestamp(int(msg["tst"]), tz=dt.timezone.utc)
+            out_dir = self.base_dir / self.user / self.device / t.strftime("%Y/%m/%d")
+            out_dir.mkdir(parents=True, exist_ok=True)
+            out_path = out_dir / f"{int(msg['tst'])}.json"
+            if self.dry_run:
+                logging.info("Would write %s", out_path)
+            else:
+                with out_path.open("w", encoding="utf-8") as f:
+                    json.dump(msg, f, separators=(",", ":"))
+
+        elif self.mode == "http":
+            if not self.http_url:
+                raise ConfigError("HTTP mode requires http_url")
+            if self.dry_run:
+                logging.info("Would POST to %s: %s", self.http_url, msg)
+            else:
+                try:
+                    resp = requests.post(
+                        self.http_url,
+                        json=msg,
+                        auth=self.http_auth,
+                        timeout=15,
+                        verify=self.http_verify,
+                        headers={"Content-Type": "application/json"},
+                    )
+                    if resp.status_code >= 400:
+                        raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:200]}")
+                except Exception as e:
+                    logging.error("HTTP POST failed: %s", e)
+                    raise
+            self._rate_limit()
+        else:
+            raise ConfigError(f"Unknown mode: {self.mode}")
+
+
+def parse_time(s: Optional[str]) -> Optional[dt.datetime]:
+    if not s:
+        return None
+    try:
+        # Allow 'Z' suffix
+        if s.endswith("Z"):
+            s = s.replace("Z", "+00:00")
+        return dt.datetime.fromisoformat(s)
+    except Exception as e:
+        raise ConfigError(f"Invalid ISO8601 time '{s}': {e}")
+
+
+def compute_speed_m_s(prev: gpxpy.gpx.GPXTrackPoint, cur: gpxpy.gpx.GPXTrackPoint) -> Optional[float]:
+    if not prev.time or not cur.time:
+        return None
+    t1 = prev.time
+    t2 = cur.time
+    if t1.tzinfo is None:
+        t1 = t1.replace(tzinfo=dt.timezone.utc)
+    if t2.tzinfo is None:
+        t2 = t2.replace(tzinfo=dt.timezone.utc)
+    dt_s = (t2 - t1).total_seconds()
+    if dt_s <= 0:
+        return 0.0
+    # Haversine distance in meters
+    R = 6371000.0
+    phi1 = math.radians(prev.latitude)
+    phi2 = math.radians(cur.latitude)
+    dphi = phi2 - phi1
+    dlambda = math.radians(cur.longitude - prev.longitude)
+    a = math.sin(dphi/2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda/2)**2
+    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
+    d_m = R * c
+    return d_m / dt_s
+
+
+def iter_gpx_points(gpx: gpxpy.gpx.GPX, start: Optional[dt.datetime], end: Optional[dt.datetime]) -> Iterable[gpxpy.gpx.GPXTrackPoint]:
+    for trk in gpx.tracks:
+        for seg in trk.segments:
+            for pt in seg.points:
+                if start and pt.time and (pt.time if pt.time.tzinfo else pt.time.replace(tzinfo=dt.timezone.utc)) < start:
+                    continue
+                if end and pt.time and (pt.time if pt.time.tzinfo else pt.time.replace(tzinfo=dt.timezone.utc)) >= end:
+                    continue
+                yield pt
+
+
+def build_arg_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        description="Convert GPX track to OwnTracks messages and output to JSONL, files, or HTTP.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    p.add_argument("gpx", help="Input .gpx file")
+    p.add_argument("--config", help="Path to YAML config")
+
+    p.add_argument("--user", help="OwnTracks user")
+    p.add_argument("--device", help="OwnTracks device")
+    p.add_argument("--tid", help="2-character tracker id (e.g. AL)")
+    p.add_argument("--acc", type=float, help="Default accuracy (meters)")
+    p.add_argument("--include-alt", dest="include_alt", action=argparse.BooleanOptionalAction, default=None,
+                   help="Include altitude if available")
+    p.add_argument("--calc-speed", dest="calc_speed", action=argparse.BooleanOptionalAction, default=None,
+                   help="Calculate speed between points (m/s)")
+    p.add_argument("--assume-timezone", help="Timezone to assume if GPX times are naive (e.g. UTC, Europe/Warsaw)")
+
+    p.add_argument("--mode", choices=["jsonl", "files", "http"], help="Output mode")
+    p.add_argument("--jsonl-path", help="Output JSONL path")
+    p.add_argument("--files-base-dir", help="Recorder store base dir for files mode")
+
+    p.add_argument("--http-url", help="Recorder /pub URL for HTTP mode")
+    p.add_argument("--http-user", help="HTTP basic auth username")
+    p.add_argument("--http-pass", help="HTTP basic auth password")
+    p.add_argument("--verify", dest="verify", action=argparse.BooleanOptionalAction, default=None,
+                   help="Verify TLS certificate in HTTP mode")
+    p.add_argument("--rate", type=float, help="Max posts per second (HTTP mode)")
+
+    p.add_argument("--start-time", help="Filter: start time ISO8601 inclusive (e.g. 2025-08-01T00:00:00Z)")
+    p.add_argument("--end-time", help="Filter: end time ISO8601 exclusive (e.g. 2025-09-01T00:00:00Z)")
+
+    p.add_argument("--dry-run", action="store_true", help="Do not write/POST, just validate and log actions")
+    p.add_argument("--log-level", default=None, help="Logging level (DEBUG, INFO, WARNING, ERROR)")
+    return p
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    args = build_arg_parser().parse_args(argv)
+
+    base_cfg = load_config(args.config)
+    cfg = merge_cli_over_config(base_cfg, args)
+
+    log_level = getattr(logging, cfg.log_level.upper(), logging.INFO)
+    logging.basicConfig(level=log_level, format="%(levelname)s %(message)s")
+
+    # Basic validations
+    in_path = Path(args.gpx)
+    if not in_path.exists():
+        logging.error("Input GPX not found: %s", in_path)
+        return 2
+
+    if cfg.mode not in {"jsonl", "files", "http"}:
+        logging.error("Invalid mode: %s", cfg.mode)
+        return 2
+
+    if cfg.mode == "jsonl" and not cfg.jsonl_path:
+        logging.error("jsonl mode requires --jsonl-path or jsonl_path in config")
+        return 2
+
+    if cfg.mode == "files" and (not cfg.files_base_dir or not cfg.user or not cfg.device):
+        logging.error("files mode requires --files-base-dir, --user, and --device")
+        return 2
+
+    if cfg.mode == "http" and not cfg.http_url:
+        logging.error("http mode requires --http-url or http.url in config")
+        return 2
+
+    if cfg.tid and len(cfg.tid) != 2:
+        logging.error("--tid must be exactly 2 characters")
+        return 2
+
+    # Parse time filters
+    start = parse_time(cfg.start_time)
+    end = parse_time(cfg.end_time)
+
+    # Load GPX
+    try:
+        with in_path.open("r", encoding="utf-8") as f:
+            gpx = gpxpy.parse(f)
+    except Exception as e:
+        logging.error("Failed to parse GPX: %s", e)
+        return 2
+
+    # Prepare sink
+    sink = OutputSink(
+        mode=cfg.mode,
+        jsonl_path=Path(cfg.jsonl_path) if cfg.jsonl_path else None,
+        base_dir=Path(cfg.files_base_dir) if cfg.files_base_dir else None,
+        http_url=cfg.http_url,
+        http_auth=(cfg.http_username, cfg.http_password) if cfg.http_username or cfg.http_password else None,
+        http_verify=cfg.http_tls_verify,
+        rate_limit_per_sec=cfg.rate_limit_per_sec,
+        user=cfg.user,
+        device=cfg.device,
+        dry_run=args.dry_run,
+    )
+
+    try:
+        sink.open()
+    except Exception as e:
+        logging.error("Failed to open output sink: %s", e)
+        return 2
+
+    # Iterate points and emit
+    count = 0
+    prev_pt: Optional[gpxpy.gpx.GPXTrackPoint] = None
+    try:
+        for pt in iter_gpx_points(gpx, start, end):
+            try:
+                vel = compute_speed_m_s(prev_pt, pt) if (cfg.calc_speed and prev_pt) else None
+                msg = OwnTracksMessage.from_point(
+                    pt,
+                    default_acc=cfg.default_acc,
+                    include_alt=cfg.include_alt,
+                    vel=vel,
+                    tid=cfg.tid,
+                )
+                sink.write(msg)
+                count += 1
+                prev_pt = pt
+            except Exception as e:
+                logging.warning("Skipping point due to error: %s", e)
+                continue
+    finally:
+        sink.close()
+
+    if count == 0:
+        logging.warning("No points exported.")
+    else:
+        logging.info("Exported %d points via mode='%s'", count, cfg.mode)
+
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except KeyboardInterrupt:
+        print("Interrupted", file=sys.stderr)
+        sys.exit(130)

gpx2owntracks.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+python gpx2owntracks.py $1 \
+  --jsonl-path test/rec/$(basename $1 | cut -d '.' -f 1).jsonl \
+  --user owntracksusr --device oskar --tid AL
+

gpx2owntracks_simple.py

@@ -0,0 +1,25 @@
+import gpxpy
+import json
+import time
+
+with open("track.gpx", "r") as f:
+    gpx = gpxpy.parse(f)
+
+points = []
+for track in gpx.tracks:
+    for segment in track.segments:
+        for point in segment.points:
+            ot_point = {
+                "_type": "location",
+                "tst": int(time.mktime(point.time.timetuple())),
+                "lat": point.latitude,
+                "lon": point.longitude,
+                "alt": point.elevation,
+                "acc": 10
+            }
+            points.append(ot_point)
+
+with open("track.json", "w") as out:
+    for p in points:
+        out.write(json.dumps(p) + "\n")
+

immich_geotag_from_owntracks.sh

@@ -0,0 +1,29 @@
+#/bin/bash
+
+# set -x 
+
+photoDir="$1"
+runCond=${2:-'no'}
+
+# Najpierw test bez zapisu:
+python3 immich_geotag_from_owntracks.py \
+  --photos $photoDir \
+  --json /home/pi/own-tracks/owntracks/store/rec/owntracksusr/oskar \
+  --json-glob "*.rec" \
+  --json-max-age-secs 1000 \
+  --recursive --dry-run
+
+if [ $runCond != 'run' ]; then
+	echo "Dry run done, exiting"
+	exit 0
+else
+	# JSON -> GPX -> geotag
+	python3 immich_geotag_from_owntracks.py \
+	  --photos $photoDir \
+	  --json /home/pi/own-tracks/owntracks/store/rec/owntracksusr/oskar/ \
+	  --recursive \
+	  --interpolate-secs 180 \
+	  --time-offset "+00:00" \
+	  --write inplace
+fi
+

immich_geotag_from_owntracks_final.sh

@@ -0,0 +1,15 @@
+#/bin/bash
+
+set -x 
+
+photoDir="$1"
+
+python3 immich_geotag_from_owntracks_final.py \
+  --photos $photoDir \
+  --json /home/pi/own-tracks/owntracks/store/rec/owntracksusr/oskar \
+  --json-glob "*.rec" \
+  --recursive --dry-run --verbose
+  
+
+
+

publich2owntracks.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+set -x 
+
+while IFS= read -r line; do
+  curl -sS -X POST \
+    -u "owntracksusr:X0Hpj5Sh0XXDTn9q-VlY1zQ5kQcHQZZO" \
+    -H 'Content-Type: application/json' \
+    --data "$line" \
+    'https://owntracks.okit.pl/pub?u=owntracksusr&d=oskar'
+done < $1

readme.md

@@ -0,0 +1,21 @@
+#1
+
+get gpx
+
+
+#2
+
+make jsonl:
+➜  immich_owntracks_enricher ./gpx2owntracks.sh test/gpx/activity_20019940630.gpx
+
+
+#3
+
+➜  immich_owntracks_enricher ./publich2owntracks.sh test/rec/activity_20019940630.jsonl 
+
+
+#4
+
+tag photos
+./immich_geotag_from_owntracks_final.sh test/photos
+