oskar/immich_owntracks
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

first commit

Browse source
This commit is contained in:
Oskar Kapala 2025-08-28 16:50:05 +02:00
commit 7f5bef7247
13 changed files with 10521 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.idea/immich_owntracks.iml Normal file
View file

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" />
<orderEntry type="jdk" jdkName="Python 3.10" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

6
.idea/inspectionProfiles/profiles_settings.xml Normal file
View file

@ -0,0 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>
</component>

7
.idea/misc.xml Normal file
View file

@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.10" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10" project-jdk-type="Python SDK" />
</project>

8
.idea/modules.xml Normal file
View file

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/immich_owntracks.iml" filepath="$PROJECT_DIR$/.idea/immich_owntracks.iml" />
</modules>
</component>
</project>

166
.idea/workspace.xml Normal file
View file

@ -0,0 +1,166 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="AutoImportSettings">
<option name="autoReloadType" value="SELECTIVE" />
</component>
<component name="ChangeListManager">
<list default="true" id="7dbbf1b8-e1f5-4e1c-be87-91d25be24c89" name="Changes" comment="" />
<option name="SHOW_DIALOG" value="false" />
<option name="HIGHLIGHT_CONFLICTS" value="true" />
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
<option name="LAST_RESOLUTION" value="IGNORE" />
</component>
<component name="FileTemplateManagerImpl">
<option name="RECENT_TEMPLATES">
<list>
<option value="Python Script" />
</list>
</option>
</component>
<component name="ProjectColorInfo"><![CDATA[{
"associatedIndex": 5
}]]></component>
<component name="ProjectId" id="31uGYCht2NS48g98eMWc2RpyoEp" />
<component name="ProjectViewState">
<option name="hideEmptyMiddlePackages" value="true" />
<option name="showLibraryContents" value="true" />
</component>
<component name="PropertiesComponent"><![CDATA[{
"keyToString": {
"ModuleVcsDetector.initialDetectionPerformed": "true",
"Python.immich_geotag_from_owntracks.executor": "Debug",
"Python.immich_geotag_from_owntracks_final.executor": "Run",
"Python.immich_geotag_from_owntracks_improoved.executor": "Run",
"RunOnceActivity.ShowReadmeOnStart": "true",
"RunOnceActivity.TerminalTabsStorage.copyFrom.TerminalArrangementManager.252": "true",
"last_opened_file_path": "/home/oskar/projects/immich_owntracks"
}
}]]></component>
<component name="RecentsManager">
<key name="CopyFile.RECENT_KEYS">
<recent name="$PROJECT_DIR$" />
</key>
</component>
<component name="RunManager" selected="Python.immich_geotag_from_owntracks_final">
<configuration name="immich_geotag_from_owntracks" type="PythonConfigurationType" factoryName="Python" nameIsGenerated="true">
<module name="immich_owntracks" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/immich_geotag_from_owntracks.py" />
<option name="PARAMETERS" value="--photos test/photos --json test/rec --json-glob &quot;*.*&quot; --json-max-age-secs 1000 --recursive --dry-run --verbose" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<configuration name="immich_geotag_from_owntracks_final" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="immich_owntracks" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/immich_geotag_from_owntracks_final.py" />
<option name="PARAMETERS" value="--photos test/photos --json test/rec --json-glob &quot;*.rec&quot; --recursive --verbose" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<configuration name="immich_geotag_from_owntracks_improoved" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
<module name="immich_owntracks" />
<option name="ENV_FILES" value="" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/immich_geotag_from_owntracks_improoved.py" />
<option name="PARAMETERS" value="--photos test/photos --json test/rec --json-glob &quot;*.rec&quot; --recursive --dry-run --verbose" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
<list>
<item itemvalue="Python.immich_geotag_from_owntracks" />
<item itemvalue="Python.immich_geotag_from_owntracks_final" />
<item itemvalue="Python.immich_geotag_from_owntracks_improoved" />
</list>
<recent_temporary>
<list>
<item itemvalue="Python.immich_geotag_from_owntracks_final" />
<item itemvalue="Python.immich_geotag_from_owntracks_improoved" />
</list>
</recent_temporary>
</component>
<component name="SharedIndexes">
<attachedChunks>
<set>
<option value="bundled-python-sdk-4ae89fc4d009-d902c0275401-com.jetbrains.pycharm.community.sharedIndexes.bundled-PC-252.25557.70" />
</set>
</attachedChunks>
</component>
<component name="TaskManager">
<task active="true" id="Default" summary="Default task">
<changelist id="7dbbf1b8-e1f5-4e1c-be87-91d25be24c89" name="Changes" comment="" />
<created>1756366589997</created>
<option name="number" value="Default" />
<option name="presentableId" value="Default" />
<updated>1756366589997</updated>
</task>
<servers />
</component>
<component name="XDebuggerManager">
<breakpoint-manager>
<breakpoints>
<line-breakpoint enabled="true" suspend="THREAD" type="python-line">
<url>file://$PROJECT_DIR$/immich_geotag_from_owntracks.py</url>
<line>216</line>
<option name="timeStamp" value="3" />
</line-breakpoint>
<line-breakpoint enabled="true" suspend="THREAD" type="python-line">
<url>file://$PROJECT_DIR$/immich_geotag_from_owntracks.py</url>
<line>96</line>
<option name="timeStamp" value="5" />
</line-breakpoint>
<line-breakpoint enabled="true" suspend="THREAD" type="python-line">
<url>file://$PROJECT_DIR$/immich_geotag_from_owntracks.py</url>
<line>86</line>
<option name="timeStamp" value="6" />
</line-breakpoint>
<line-breakpoint enabled="true" suspend="THREAD" type="python-line">
<url>file://$PROJECT_DIR$/immich_geotag_from_owntracks.py</url>
<line>87</line>
<option name="timeStamp" value="7" />
</line-breakpoint>
</breakpoints>
</breakpoint-manager>
</component>
</project>

24
gpw2owntrtacks_simple.py Normal file
View file

@ -0,0 +1,24 @@
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")

255
immich_geotag_from_owntracks.py Normal file
View file

@ -0,0 +1,255 @@
#!/usr/bin/env python3
"""
immich_geotag_from_owntracks.py
Geotaguj zdjęcia na podstawie śladu z OwnTracks (JSON) lub pliku GPX
z wykorzystaniem exiftool. Idealne do późniejszego importu / reindeksu w Immich.
Wymagania:
- Python 3.9+
- exiftool (musi być w PATH, np. `apt-get install exiftool` lub pakiet systemowy)
- (Opcjonalnie) uprawnienia do zapisu plików JPG/HEIC/PNG itp.
Przykłady:
1) JSON -> GPX -> geotag (rekurencyjnie po katalogu ze zdjęciami):
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/owntracks_json \
--json-max-age-secs 10 \
--interpolate-secs 180 \
--time-offset "+00:00" \
--write inplace
2) GPX -> geotag:
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--gpx /ścieżka/track.gpx \
--interpolate-secs 180 \
--time-offset "+00:00" \
--write inplace
Po zakończeniu możesz wykonać reindeks w Immich, aby wczytać nowe EXIF GPS.
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any, Optional
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Geotaguj zdjęcia na podstawie OwnTracks JSON lub GPX (dla Immich).")
g_input = p.add_argument_group("Wejście śladu")
g_input.add_argument("--gpx", type=str, help="Ścieżka do pliku GPX (opcjonalnie zamiast --json).")
g_input.add_argument("--json", type=str, help="Ścieżka do pliku JSON OwnTracks lub katalogu z JSON-ami.")
g_input.add_argument("--json-glob", type=str, default="*.json", help="Wzorzec plików JSON gdy --json wskazuje katalog (domyślnie: *.json).")
g_match = p.add_argument_group("Parametry dopasowania czasu")
g_match.add_argument("--interpolate-secs", type=int, default=180, help="Maksymalne okno interpolacji (GeoMaxIntSecs) w sekundach (domyślnie 180).")
g_match.add_argument("--json-max-age-secs", type=int, default=10, help="Ignoruj punkty JSON bez timestampu lub dane starsze/niepoprawne (bufor bezpieczeństwa).")
g_match.add_argument("--time-offset", type=str, default="+00:00",
help="Korekta czasu EXIF względem GPX, np. '+00:00', '+01:00', '-00:30'. Przekazywana do exiftool jako -geosync.")
g_photos = p.add_argument_group("Zdjęcia")
g_photos.add_argument("--photos", type=str, required=True, help="Katalog lub plik ze zdjęciem(i).")
g_photos.add_argument("--recursive", action="store_true", help="Przetwarzaj rekursywnie katalog zdjęć (-r dla exiftool).")
g_out = p.add_argument_group("Zapis")
g_out.add_argument("--write", choices=["inplace", "backup"], default="inplace",
help="Tryb zapisu: 'inplace' (overwrite_original) lub 'backup' (domyślny tryb exiftool z plikami *_original).")
g_out.add_argument("--dry-run", action="store_true", help="Nie zapisuj EXIF, tylko pokaż co byłoby wykonane.")
g_misc = p.add_argument_group("Różne")
g_misc.add_argument("--tmpdir", type=str, help="Katalog tymczasowy na plik GPX (jeśli generowany z JSON).")
g_misc.add_argument("--verbose", action="store_true", help="Więcej logów.")
return p.parse_args()
def discover_json_files(path: Path, pattern: str) -> List[Path]:
if path.is_file():
return [path]
elif path.is_dir():
return sorted(path.rglob(pattern))
else:
raise FileNotFoundError(f"Ścieżka JSON nie istnieje: {path}")
def load_owntracks_points(json_files: List[Path], max_age_seconds: int, verbose=False) -> List[Dict[str, Any]]:
"""
Wczytuje punkty OwnTracks z plików JSON.
Oczekuje kluczy: lat, lon, tst (epoch sekundy). Toleruje inne pola.
Zwraca listę posortowaną po czasie (UTC).
"""
points = []
for f in json_files:
try:
with open(f, "r", encoding="utf-8") as fh:
data = json.load(fh)
except Exception as e:
if verbose:
print(f"[WARN] Nie udało się wczytać {f}: {e}", file=sys.stderr)
continue
# Obsługa: plik zawiera pojedynczy obiekt lub listę obiektów
entries = data if isinstance(data, list) else [data]
for entry in entries:
try:
lat = float(entry["lat"])
lon = float(entry["lon"])
tst = int(entry["tst"]) # epoch (sekundy)
if tst <= 0:
continue
dt = datetime.fromtimestamp(tst, tz=timezone.utc)
points.append({"lat": lat, "lon": lon, "time": dt})
except Exception:
continue
points.sort(key=lambda x: x["time"])
if verbose:
if points:
print(f"[INFO] Załadowano {len(points)} punktów OwnTracks. Zakres: {points[0]['time']} .. {points[-1]['time']} (UTC)")
else:
print("[WARN] Nie znaleziono żadnych punktów OwnTracks w JSON.")
return points
def write_gpx(points: List[Dict[str, Any]], out_path: Path, verbose=False) -> None:
"""
Minimalny zapis GPX 1.1 z jedną trasą (trk) i jednym odcinkiem (trkseg).
"""
def iso8601(dt: datetime) -> str:
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
with open(out_path, "w", encoding="utf-8") as gpx:
gpx.write('<?xml version="1.0" encoding="UTF-8"?>\n')
gpx.write('<gpx version="1.1" creator="immich_geotag_from_owntracks.py" xmlns="http://www.topografix.com/GPX/1/1">\n')
gpx.write(' <trk>\n')
gpx.write(' <name>OwnTracks Export</name>\n')
gpx.write(' <trkseg>\n')
for pt in points:
gpx.write(f' <trkpt lat="{pt["lat"]:.8f}" lon="{pt["lon"]:.8f}"><time>{iso8601(pt["time"])}</time></trkpt>\n')
gpx.write(' </trkseg>\n')
gpx.write(' </trk>\n')
gpx.write('</gpx>\n')
if verbose:
print(f"[INFO] Zapisano GPX: {out_path} ({len(points)} punktów)")
def ensure_exiftool_available() -> None:
try:
subprocess.run(["exiftool", "-ver"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except (subprocess.CalledProcessError, FileNotFoundError):
print("[ERROR] exiftool nie jest dostępny w PATH. Zainstaluj exiftool i spróbuj ponownie.", file=sys.stderr)
sys.exit(2)
def build_exiftool_cmd(gpx_path: Path, photos_path: Path, interpolate_secs: int, time_offset: str,
recursive: bool, write_mode: str, dry_run: bool) -> List[str]:
cmd = ["exiftool"]
if recursive and photos_path.is_dir():
cmd.append("-r")
cmd.append("-api")
cmd.append(f"GeoMaxIntSecs={interpolate_secs}")
if time_offset:
cmd.append(f"-geosync={time_offset}")
cmd.append("-geotag")
cmd.append(str(gpx_path))
if write_mode == "inplace" and not dry_run:
cmd.append("-overwrite_original")
cmd.append(str(photos_path))
if dry_run:
cmd.insert(1, "-n")
cmd.append("-S")
cmd.append("-v2")
return cmd
def run_exiftool(cmd: List[str], verbose=False) -> int:
if verbose:
print("[INFO] Uruchamiam:", " ".join(cmd))
try:
proc = subprocess.run(cmd, check=False)
return proc.returncode
except FileNotFoundError:
print("[ERROR] exiftool nie znaleziony.", file=sys.stderr)
return 127
def main() -> int:
args = parse_args()
ensure_exiftool_available()
photos_path = Path(args.photos)
if not photos_path.exists():
print(f"[ERROR] Ścieżka do zdjęć nie istnieje: {photos_path}", file=sys.stderr)
return 1
gpx_path: Optional[Path] = None
temp_dir = None
if args.gpx:
gpx_path = Path(args.gpx)
if not gpx_path.exists():
print(f"[ERROR] Plik GPX nie istnieje: {gpx_path}", file=sys.stderr)
return 1
if args.verbose:
print(f"[INFO] Używam dostarczonego GPX: {gpx_path}")
else:
if not args.json:
print("[ERROR] Musisz podać --gpx lub --json.", file=sys.stderr)
return 1
json_path = Path(args.json)
try:
files = discover_json_files(json_path, args.json_glob)
except FileNotFoundError as e:
print(f"[ERROR] {e}", file=sys.stderr)
return 1
points = load_owntracks_points(files, args.json_max_age_secs, verbose=args.verbose)
if not points:
print("[ERROR] Brak punktów z OwnTracks po wczytaniu JSON.", file=sys.stderr)
return 1
temp_dir = Path(args.tmpdir) if args.tmpdir else Path(tempfile.mkdtemp(prefix="owntracks_gpx_"))
temp_dir.mkdir(parents=True, exist_ok=True)
gpx_path = temp_dir / "owntracks_export.gpx"
write_gpx(points, gpx_path, verbose=args.verbose)
assert gpx_path is not None
cmd = build_exiftool_cmd(
gpx_path=gpx_path,
photos_path=photos_path,
interpolate_secs=args.interpolate_secs,
time_offset=args.time_offset,
recursive=args.recursive,
write_mode=args.write,
dry_run=args.dry_run,
)
rc = run_exiftool(cmd, verbose=args.verbose)
if rc != 0:
print(f"[ERROR] exiftool zakończył się kodem {rc}.", file=sys.stderr)
return rc
if args.dry_run:
print("[OK] Dry-run zakończony. Wygląda dobrze. Usuń --dry-run, aby zapisać EXIF.")
else:
print("[OK] Geotagowanie zakończone sukcesem. Uruchom reindeks w Immich, aby wczytać GPS.")
if temp_dir:
print(f"[INFO] Tymczasowy GPX: {gpx_path} (możesz go zachować do archiwum).")
return 0
if __name__ == "__main__":
sys.exit(main())

329
immich_geotag_from_owntracks_final.py Normal file
View file

@ -0,0 +1,329 @@
#!/usr/bin/env python3
"""
immich_geotag_from_owntracks.py
Geotaguj zdjęcia na podstawie śladu z OwnTracks (JSON/JSONL/NDJSON/.rec) lub pliku GPX
z wykorzystaniem exiftool. Idealne do późniejszego importu / reindeksu w Immich.
Nowości:
- Obsługa JSON Lines / NDJSON oraz „sklejonych” obiektów JSON.
- Domyślne przeszukiwanie katalogu: *.json, *.jsonl, *.ndjson, *.rec
- **Bezpiecznik:** domyślnie NIE nadpisuje GPS, jeśli już istnieje w EXIF.
(użyj --overwrite-existing-gps aby nadpisać)
Wymagania:
- Python 3.9+
- exiftool (musi być w PATH, np. `apt-get install exiftool`)
Przykłady:
1) Katalog OwnTracks Recorder: owntracks/store/rec (pliki .rec):
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/owntracks/store/rec \
--recursive \
--write inplace
2) JSON/JSONL/NDJSON -> GPX -> geotag (bez nadpisywania istniejącego GPS):
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/logów \
--interpolate-secs 180 \
--time-offset "+00:00" \
--write inplace
3) Wymuś nadpisywanie istniejącego GPS:
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/logów \
--overwrite-existing-gps \
--write inplace
"""
import argparse
import json
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any, Optional, Iterable
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Geotaguj zdjęcia na podstawie OwnTracks JSON/JSONL/NDJSON/.rec lub GPX (dla Immich).")
g_input = p.add_argument_group("Wejście śladu")
g_input.add_argument("--gpx", type=str, help="Ścieżka do pliku GPX (opcjonalnie zamiast --json).")
g_input.add_argument("--json", type=str, help="Ścieżka do pliku/katalogu OwnTracks (JSON/JSONL/NDJSON/.rec).")
g_input.add_argument(
"--json-glob",
type=str,
default="*.json,*.jsonl,*.ndjson,*.rec",
help="Wzorce plików rozdzielone przecinkami gdy --json wskazuje katalog (domyślnie: *.json,*.jsonl,*.ndjson,*.rec)."
)
g_match = p.add_argument_group("Parametry dopasowania czasu")
g_match.add_argument("--interpolate-secs", type=int, default=180, help="Maksymalne okno interpolacji (GeoMaxIntSecs) w sekundach (domyślnie 180).")
g_match.add_argument("--time-offset", type=str, default="+00:00",
help="Korekta czasu EXIF względem GPX, np. '+00:00', '+01:00', '-00:30'. Przekazywana do exiftool jako -geosync.")
g_photos = p.add_argument_group("Zdjęcia")
g_photos.add_argument("--photos", type=str, required=True, help="Katalog lub plik ze zdjęciem(i).")
g_photos.add_argument("--recursive", action="store_true", help="Przetwarzaj rekursywnie katalog zdjęć (-r dla exiftool).")
g_out = p.add_argument_group("Zapis")
g_out.add_argument("--write", choices=["inplace", "backup"], default="inplace",
help="Tryb zapisu: 'inplace' (overwrite_original) lub 'backup' (domyślny tryb exiftool z plikami *_original).")
g_out.add_argument("--dry-run", action="store_true", help="Nie zapisuj EXIF, tylko pokaż co byłoby wykonane.")
g_misc = p.add_argument_group("Różne")
g_misc.add_argument("--tmpdir", type=str, help="Katalog tymczasowy na plik GPX (jeśli generowany z JSON).")
g_misc.add_argument("--verbose", action="store_true", help="Więcej logów.")
g_misc.add_argument("--overwrite-existing-gps", action="store_true",
help="Domyślnie nie nadpisuje istniejącego GPS w EXIF. Dodaj ten przełącznik, aby nadpisać.")
return p.parse_args()
def _glob_patterns(directory: Path, patterns_csv: str) -> List[Path]:
patterns = [p.strip() for p in patterns_csv.split(",") if p.strip()]
files: List[Path] = []
for patt in patterns:
files.extend(directory.rglob(patt))
return sorted(set(files))
def discover_json_files(path: Path, patterns_csv: str) -> List[Path]:
if path.is_file():
return [path]
elif path.is_dir():
return _glob_patterns(path, patterns_csv)
else:
raise FileNotFoundError(f"Ścieżka z danymi OwnTracks nie istnieje: {path}")
def _iter_json_objects_from_string(buf: str) -> Iterable[Any]:
"""Próbuje: klasyczny JSON -> JSONL/NDJSON linia po linii -> sekwencyjne raw_decode."""
# 1) klasyczny JSON
try:
obj = json.loads(buf)
if isinstance(obj, list):
for x in obj:
yield x
return
else:
yield obj
return
except Exception:
pass
# 2) JSON Lines / NDJSON
parsed_any = False
for line in buf.splitlines():
s = line.strip()
if not s or s.startswith("#"):
continue
try:
yield json.loads(s)
parsed_any = True
except Exception:
pass
if parsed_any:
return
# 3) Wiele sklejonych obiektów JSON
dec = json.JSONDecoder()
i = 0
n = len(buf)
while i < n:
while i < n and buf[i].isspace():
i += 1
if i >= n:
break
try:
obj, j = dec.raw_decode(buf, idx=i)
yield obj
i = j
except Exception:
i += 1
def load_owntracks_points(json_files: List[Path], verbose=False) -> List[Dict[str, Any]]:
"""Czyta punkty OwnTracks z wielu wariantów plików. Wymaga lat/lon/tst."""
points: List[Dict[str, Any]] = []
for f in json_files:
try:
raw = f.read_text(encoding="utf-8")
except Exception as e:
if verbose:
print(f"[WARN] Nie udało się odczytać {f}: {e}", file=sys.stderr)
continue
any_in_file = False
for entry in _iter_json_objects_from_string(raw):
try:
lat = float(entry["lat"])
lon = float(entry["lon"])
tst = int(entry["tst"]) # epoch sekundy
if tst <= 0:
continue
dt = datetime.fromtimestamp(tst, tz=timezone.utc)
points.append({"lat": lat, "lon": lon, "time": dt})
any_in_file = True
except Exception:
continue
if verbose and not any_in_file:
print(f"[WARN] Brak poprawnych rekordów lat/lon/tst w pliku: {f}", file=sys.stderr)
points.sort(key=lambda x: x["time"])
if verbose:
if points:
print(f"[INFO] Załadowano {len(points)} punktów OwnTracks. Zakres: {points[0]['time']} .. {points[-1]['time']} (UTC)")
else:
print("[WARN] Nie znaleziono żadnych punktów OwnTracks (lat/lon/tst).")
return points
def write_gpx(points: List[Dict[str, Any]], out_path: Path, verbose=False) -> None:
def iso8601(dt: datetime) -> str:
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
with open(out_path, "w", encoding="utf-8") as gpx:
gpx.write('<?xml version="1.0" encoding="UTF-8"?>\n')
gpx.write('<gpx version="1.1" creator="immich_geotag_from_owntracks.py" xmlns="http://www.topografix.com/GPX/1/1">\n')
gpx.write(' <trk>\n')
gpx.write(' <name>OwnTracks Export</name>\n')
gpx.write(' <trkseg>\n')
for pt in points:
gpx.write(f' <trkpt lat="{pt["lat"]:.8f}" lon="{pt["lon"]:.8f}"><time>{iso8601(pt["time"])}</time></trkpt>\n')
gpx.write(' </trkseg>\n')
gpx.write(' </trk>\n')
gpx.write('</gpx>\n')
if verbose:
print(f"[INFO] Zapisano GPX: {out_path} ({len(points)} punktów)")
def ensure_exiftool_available() -> None:
try:
subprocess.run(["exiftool", "-ver"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except (subprocess.CalledProcessError, FileNotFoundError):
print("[ERROR] exiftool nie jest dostępny w PATH. Zainstaluj exiftool i spróbuj ponownie.", file=sys.stderr)
sys.exit(2)
def build_exiftool_cmd(
gpx_path: Path,
photos_path: Path,
interpolate_secs: int,
time_offset: str,
recursive: bool,
write_mode: str,
dry_run: bool,
skip_existing_gps: bool,
) -> List[str]:
cmd: List[str] = ["exiftool"]
if recursive and photos_path.is_dir():
cmd.append("-r")
# Pomijaj pliki, które już mają GPS
if skip_existing_gps:
cmd.extend(["-if", "not $GPSLatitude"])
# dopasowanie w oknie czasowym
cmd.extend(["-api", f"GeoMaxIntSecs={interpolate_secs}"])
# korekta czasu
if time_offset:
cmd.append(f"-geosync={time_offset}")
# źródło geotagów
cmd.extend(["-geotag", str(gpx_path)])
# zapis
if write_mode == "inplace" and not dry_run:
cmd.append("-overwrite_original")
# cel (ostatni argument)
cmd.append(str(photos_path))
# symulacja
if dry_run:
cmd.insert(1, "-n") # wartości numeryczne bez formatowania
cmd.append("-S") # krótsze wyjście
cmd.append("-v2") # szczegóły operacji
return cmd
def run_exiftool(cmd: List[str], verbose=False) -> int:
if verbose:
print("[INFO] Uruchamiam:", " ".join(cmd))
try:
proc = subprocess.run(cmd, check=False)
return proc.returncode
except FileNotFoundError:
print("[ERROR] exiftool nie znaleziony.", file=sys.stderr)
return 127
def main() -> int:
args = parse_args()
ensure_exiftool_available()
photos_path = Path(args.photos)
if not photos_path.exists():
print(f"[ERROR] Ścieżka do zdjęć nie istnieje: {photos_path}", file=sys.stderr)
return 1
# 1) Przygotuj GPX
gpx_path: Optional[Path] = None
temp_dir: Optional[Path] = None
if args.gpx:
gpx_path = Path(args.gpx)
if not gpx_path.exists():
print(f"[ERROR] Plik GPX nie istnieje: {gpx_path}", file=sys.stderr)
return 1
if args.verbose:
print(f"[INFO] Używam dostarczonego GPX: {gpx_path}")
else:
if not args.json:
print("[ERROR] Musisz podać --gpx lub --json.", file=sys.stderr)
return 1
json_path = Path(args.json)
try:
files = discover_json_files(json_path, args.json_glob)
except FileNotFoundError as e:
print(f"[ERROR] {e}", file=sys.stderr)
return 1
if args.verbose:
print(f"[INFO] Znaleziono {len(files)} plików danych (wzorce: {args.json_glob}).")
points = load_owntracks_points(files, verbose=args.verbose)
if not points:
print("[ERROR] Brak punktów z OwnTracks po wczytaniu plików (lat/lon/tst nie znaleziono).", file=sys.stderr)
return 1
temp_dir = Path(args.tmpdir) if args.tmpdir else Path(tempfile.mkdtemp(prefix="owntracks_gpx_"))
temp_dir.mkdir(parents=True, exist_ok=True)
gpx_path = temp_dir / "owntracks_export.gpx"
write_gpx(points, gpx_path, verbose=args.verbose)
assert gpx_path is not None
# 2) Zbuduj i uruchom exiftool
skip_existing_gps = not args.overwrite_existing_gps
cmd = build_exiftool_cmd(
gpx_path=gpx_path,
photos_path=photos_path,
interpolate_secs=args.interpolate_secs,
time_offset=args.time_offset,
recursive=args.recursive,
write_mode=args.write,
dry_run=args.dry_run,
skip_existing_gps=skip_existing_gps,
)
rc = run_exiftool(cmd, verbose=args.verbose)
if rc != 0:
print(f"[ERROR] exiftool zakończył się kodem {rc}.", file=sys.stderr)
return rc
if args.dry_run:
print("[OK] Dry-run zakończony. Wygląda dobrze. Usuń --dry-run, aby zapisać EXIF.")
else:
print("[OK] Geotagowanie zakończone sukcesem. (Pliki z istniejącym GPS zostały pominięte.)")
if temp_dir:
print(f"[INFO] Tymczasowy GPX: {gpx_path} (możesz go zachować do archiwum).")
return 0
if __name__ == "__main__":
sys.exit(main())

344
immich_geotag_from_owntracks_improoved.py Normal file
View file

@ -0,0 +1,344 @@
#!/usr/bin/env python3
"""
immich_geotag_from_owntracks.py
Geotaguj zdjęcia na podstawie śladu z OwnTracks (JSON/JSONL/NDJSON/.rec) lub pliku GPX
z wykorzystaniem exiftool. Idealne do późniejszego importu / reindeksu w Immich.
Nowości w tej wersji:
- Obsługa JSON Lines / NDJSON oraz „sklejonych” obiektów JSON.
- Domyślne przeszukiwanie katalogu: *.json, *.jsonl, *.ndjson, *.rec
Wymagania:
- Python 3.9+
- exiftool (musi być w PATH, np. `apt-get install exiftool` lub pakiet systemowy)
Przykłady:
1) Katalog OwnTracks Recorder: owntracks/store/rec (pliki .rec):
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/owntracks/store/rec \
--recursive \
--write inplace
2) JSON/JSONL/NDJSON -> GPX -> geotag:
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--json /ścieżka/do/logów \
--interpolate-secs 180 \
--time-offset "+00:00" \
--write inplace
3) GPX -> geotag:
python3 immich_geotag_from_owntracks.py \
--photos /ścieżka/do/zdjęć \
--gpx /ścieżka/track.gpx \
--interpolate-secs 180 \
--time-offset "+00:00" \
--write inplace
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Dict, Any, Optional, Iterable
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Geotaguj zdjęcia na podstawie OwnTracks JSON/JSONL/NDJSON/.rec lub GPX (dla Immich).")
g_input = p.add_argument_group("Wejście śladu")
g_input.add_argument("--gpx", type=str, help="Ścieżka do pliku GPX (opcjonalnie zamiast --json).")
g_input.add_argument("--json", type=str, help="Ścieżka do pliku/katalogu OwnTracks (JSON/JSONL/NDJSON/.rec).")
g_input.add_argument(
"--json-glob",
type=str,
default="*.json,*.jsonl,*.ndjson,*.rec",
help="Wzorce plików rozdzielone przecinkami gdy --json wskazuje katalog (domyślnie: *.json,*.jsonl,*.ndjson,*.rec)."
)
g_match = p.add_argument_group("Parametry dopasowania czasu")
g_match.add_argument("--interpolate-secs", type=int, default=180, help="Maksymalne okno interpolacji (GeoMaxIntSecs) w sekundach (domyślnie 180).")
g_match.add_argument("--json-max-age-secs", type=int, default=10, help="(Nieużywane obecnie) Bufor sanity-check; zostaw domyślnie.")
g_match.add_argument("--time-offset", type=str, default="+00:00",
help="Korekta czasu EXIF względem GPX, np. '+00:00', '+01:00', '-00:30'. Przekazywana do exiftool jako -geosync.")
g_photos = p.add_argument_group("Zdjęcia")
g_photos.add_argument("--photos", type=str, required=True, help="Katalog lub plik ze zdjęciem(i).")
g_photos.add_argument("--recursive", action="store_true", help="Przetwarzaj rekursywnie katalog zdjęć (-r dla exiftool).")
g_out = p.add_argument_group("Zapis")
g_out.add_argument("--write", choices=["inplace", "backup"], default="inplace",
help="Tryb zapisu: 'inplace' (overwrite_original) lub 'backup' (domyślny tryb exiftool z plikami *_original).")
g_out.add_argument("--dry-run", action="store_true", help="Nie zapisuj EXIF, tylko pokaż co byłoby wykonane.")
g_misc = p.add_argument_group("Różne")
g_misc.add_argument("--tmpdir", type=str, help="Katalog tymczasowy na plik GPX (jeśli generowany z JSON).")
g_misc.add_argument("--verbose", action="store_true", help="Więcej logów.")
return p.parse_args()
def _glob_patterns(directory: Path, patterns_csv: str) -> List[Path]:
patterns = [p.strip() for p in patterns_csv.split(",") if p.strip()]
files: List[Path] = []
for patt in patterns:
files.extend(directory.rglob(patt))
# deduplikacja i sort
return sorted(set(files))
def discover_json_files(path: Path, patterns_csv: str) -> List[Path]:
if path.is_file():
return [path]
elif path.is_dir():
return _glob_patterns(path, patterns_csv)
else:
raise FileNotFoundError(f"Ścieżka z danymi OwnTracks nie istnieje: {path}")
def _iter_json_objects_from_string(buf: str) -> Iterable[Any]:
"""
Parsuje wiele obiektów JSON z jednego stringa:
- najpierw próbuje json.loads (pojedynczy obiekt lub lista),
- jeśli się nie uda, próbuje JSON Lines/NDJSON (po linii),
- jeśli dalej nic, używa sekwencyjnego dekodera raw_decode, by wyciągnąć sklejone obiekty.
"""
# 1) klasyczny JSON
try:
obj = json.loads(buf)
if isinstance(obj, list):
for x in obj:
yield x
return
else:
yield obj
return
except Exception:
pass
# 2) JSON Lines (NDJSON) — po jednej strukturze na linię
parsed_any = False
for line in buf.splitlines():
s = line.strip()
if not s:
continue
# pomijaj linie typu komentarz (rzadko spotykane)
if s.startswith("#"):
continue
try:
yield json.loads(s)
parsed_any = True
except Exception:
# ignoruj linie, które nie są JSON-em; spróbujemy trybu 3
pass
if parsed_any:
return
# 3) Wiele sklejonych obiektów JSON w jednym strumieniu
dec = json.JSONDecoder()
i = 0
n = len(buf)
while i < n:
# pomiń białe znaki
while i < n and buf[i].isspace():
i += 1
if i >= n:
break
try:
obj, j = dec.raw_decode(buf, idx=i)
yield obj
i = j
except Exception:
# jeśli nie da się odczytać, przesuń o jeden znak i próbuj dalej
i += 1
def load_owntracks_points(json_files: List[Path], verbose=False) -> List[Dict[str, Any]]:
"""
Wczytuje punkty OwnTracks z plików w formatach:
- JSON (obiekt lub lista),
- JSON Lines / NDJSON,
- wiele sklejonych obiektów JSON,
- .rec (jak wyżej — faktycznie JSON/NDJSON).
Oczekuje, że każdy obiekt ma klucze: lat, lon, tst (epoch sekundy).
Zwraca listę posortowaną po czasie (UTC).
"""
points: List[Dict[str, Any]] = []
for f in json_files:
try:
raw = f.read_text(encoding="utf-8")
except Exception as e:
if verbose:
print(f"[WARN] Nie udało się odczytać {f}: {e}", file=sys.stderr)
continue
any_in_file = False
for entry in _iter_json_objects_from_string(raw):
try:
# niektóre rekordy OwnTracks mogą mieć pola pod innymi kluczami, np. "lat", "lon", "tst"
lat = float(entry["lat"])
lon = float(entry["lon"])
tst = int(entry["tst"]) # epoch sekundy
if tst <= 0:
continue
dt = datetime.fromtimestamp(tst, tz=timezone.utc)
points.append({"lat": lat, "lon": lon, "time": dt})
any_in_file = True
except Exception:
# brak wymaganych pól/nie ten rekord (np. wpisy activity/waypoints)
continue
if verbose and not any_in_file:
print(f"[WARN] Brak poprawnych rekordów lat/lon/tst w pliku: {f}", file=sys.stderr)
points.sort(key=lambda x: x["time"])
if verbose:
if points:
print(f"[INFO] Załadowano {len(points)} punktów OwnTracks. Zakres: {points[0]['time']} .. {points[-1]['time']} (UTC)")
else:
print("[WARN] Nie znaleziono żadnych punktów OwnTracks (lat/lon/tst).")
return points
def write_gpx(points: List[Dict[str, Any]], out_path: Path, verbose=False) -> None:
def iso8601(dt: datetime) -> str:
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
with open(out_path, "w", encoding="utf-8") as gpx:
gpx.write('<?xml version="1.0" encoding="UTF-8"?>\n')
gpx.write('<gpx version="1.1" creator="immich_geotag_from_owntracks.py" xmlns="http://www.topografix.com/GPX/1/1">\n')
gpx.write(' <trk>\n')
gpx.write(' <name>OwnTracks Export</name>\n')
gpx.write(' <trkseg>\n')
for pt in points:
gpx.write(f' <trkpt lat="{pt["lat"]:.8f}" lon="{pt["lon"]:.8f}"><time>{iso8601(pt["time"])}</time></trkpt>\n')
gpx.write(' </trkseg>\n')
gpx.write(' </trk>\n')
gpx.write('</gpx>\n')
if verbose:
print(f"[INFO] Zapisano GPX: {out_path} ({len(points)} punktów)")
def ensure_exiftool_available() -> None:
try:
subprocess.run(["exiftool", "-ver"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except (subprocess.CalledProcessError, FileNotFoundError):
print("[ERROR] exiftool nie jest dostępny w PATH. Zainstaluj exiftool i spróbuj ponownie.", file=sys.stderr)
sys.exit(2)
def build_exiftool_cmd(gpx_path: Path, photos_path: Path, interpolate_secs: int, time_offset: str,
recursive: bool, write_mode: str, dry_run: bool) -> List[str]:
cmd = ["exiftool"]
if recursive and photos_path.is_dir():
cmd.append("-r")
cmd.append("-api")
cmd.append(f"GeoMaxIntSecs={interpolate_secs}")
if time_offset:
cmd.append(f"-geosync={time_offset}")
cmd.append("-geotag")
cmd.append(str(gpx_path))
if write_mode == "inplace" and not dry_run:
cmd.append("-overwrite_original")
cmd.append(str(photos_path))
if dry_run:
cmd.insert(1, "-n") # numery bez formatowania
cmd.append("-S")
cmd.append("-v2")
return cmd
def run_exiftool(cmd: List[str], verbose=False) -> int:
if verbose:
print("[INFO] Uruchamiam:", " ".join(cmd))
try:
proc = subprocess.run(cmd, check=False)
return proc.returncode
except FileNotFoundError:
print("[ERROR] exiftool nie znaleziony.", file=sys.stderr)
return 127
def main() -> int:
args = parse_args()
ensure_exiftool_available()
photos_path = Path(args.photos)
if not photos_path.exists():
print(f"[ERROR] Ścieżka do zdjęć nie istnieje: {photos_path}", file=sys.stderr)
return 1
# 1) Przygotuj GPX
gpx_path: Optional[Path] = None
temp_dir: Optional[Path] = None
if args.gpx:
gpx_path = Path(args.gpx)
if not gpx_path.exists():
print(f"[ERROR] Plik GPX nie istnieje: {gpx_path}", file=sys.stderr)
return 1
if args.verbose:
print(f"[INFO] Używam dostarczonego GPX: {gpx_path}")
else:
if not args.json:
print("[ERROR] Musisz podać --gpx lub --json.", file=sys.stderr)
return 1
json_path = Path(args.json)
try:
files = discover_json_files(json_path, args.json_glob)
except FileNotFoundError as e:
print(f"[ERROR] {e}", file=sys.stderr)
return 1
if args.verbose:
print(f"[INFO] Znaleziono {len(files)} plików danych (wzorce: {args.json_glob}).")
points = load_owntracks_points(files, verbose=args.verbose)
if not points:
print("[ERROR] Brak punktów z OwnTracks po wczytaniu plików (lat/lon/tst nie znaleziono).", file=sys.stderr)
return 1
temp_dir = Path(args.tmpdir) if args.tmpdir else Path(tempfile.mkdtemp(prefix="owntracks_gpx_"))
temp_dir.mkdir(parents=True, exist_ok=True)
gpx_path = temp_dir / "owntracks_export.gpx"
write_gpx(points, gpx_path, verbose=args.verbose)
assert gpx_path is not None
# 2) Zbuduj i uruchom exiftool
cmd = build_exiftool_cmd(
gpx_path=gpx_path,
photos_path=photos_path,
interpolate_secs=args.interpolate_secs,
time_offset=args.time_offset,
recursive=args.recursive,
write_mode=args.write,
dry_run=args.dry_run,
)
rc = run_exiftool(cmd, verbose=args.verbose)
if rc != 0:
print(f"[ERROR] exiftool zakończył się kodem {rc}.", file=sys.stderr)
return rc
if args.dry_run:
print("[OK] Dry-run zakończony. Wygląda dobrze. Usuń --dry-run, aby zapisać EXIF.")
else:
print("[OK] Geotagowanie zakończone sukcesem. Uruchom reindeks w Immich, aby wczytać GPS.")
if temp_dir:
print(f"[INFO] Tymczasowy GPX: {gpx_path} (możesz go zachować do archiwum).")
return 0
if __name__ == "__main__":
sys.exit(main())

BIN
test/photos/5c121efc-b8b0-42e0-b985-4b2c577fe26a.jpg Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 9.3 MiB

BIN
test/photos/64141f44-a825-493a-b9ea-54878de56f03.jpg Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 10 MiB

8060
test/rec/2025-08.rec Normal file
View file

File diff suppressed because it is too large Load diff

1314
test/rec/activity_20019940630.jsonl Normal file
View file

File diff suppressed because it is too large Load diff