Improve GPS reverse geocoding
This commit is contained in:
parent
bc935d2e0a
commit
e35201496f
2 changed files with 181 additions and 16 deletions
185
movmaker.py
185
movmaker.py
|
|
@ -20,6 +20,8 @@ import subprocess
|
|||
import sys
|
||||
import tempfile
|
||||
import unicodedata
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
|
@ -117,12 +119,10 @@ def ffprobe_creation_date(path: Path) -> datetime | None:
|
|||
return None
|
||||
|
||||
|
||||
def exif_image_date(path: Path) -> datetime | None:
|
||||
"""Read common JPEG/TIFF EXIF date tags using only the stdlib."""
|
||||
def image_tiff_data(path: Path) -> bytes | None:
|
||||
data = path.read_bytes()[:1024 * 1024]
|
||||
if data.startswith(b"\xff\xd8"):
|
||||
pos = 2
|
||||
tiff: bytes | None = None
|
||||
while pos + 4 <= len(data):
|
||||
if data[pos] != 0xFF:
|
||||
break
|
||||
|
|
@ -136,12 +136,16 @@ def exif_image_date(path: Path) -> datetime | None:
|
|||
segment = data[pos + 2:pos + size]
|
||||
pos += size
|
||||
if marker == 0xE1 and segment.startswith(b"Exif\x00\x00"):
|
||||
tiff = segment[6:]
|
||||
break
|
||||
if tiff is None:
|
||||
return None
|
||||
else:
|
||||
tiff = data
|
||||
return segment[6:]
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def exif_image_date(path: Path) -> datetime | None:
|
||||
"""Read common JPEG/TIFF EXIF date tags using only the stdlib."""
|
||||
tiff = image_tiff_data(path)
|
||||
if tiff is None:
|
||||
return None
|
||||
|
||||
if tiff[:2] == b"II":
|
||||
endian = "little"
|
||||
|
|
@ -193,6 +197,163 @@ def exif_image_date(path: Path) -> datetime | None:
|
|||
return None
|
||||
|
||||
|
||||
def exif_image_gps(path: Path) -> tuple[float, float] | None:
|
||||
tiff = image_tiff_data(path)
|
||||
if tiff is None:
|
||||
return None
|
||||
if tiff[:2] == b"II":
|
||||
endian = "little"
|
||||
elif tiff[:2] == b"MM":
|
||||
endian = "big"
|
||||
else:
|
||||
return None
|
||||
|
||||
def u16(off: int) -> int:
|
||||
return int.from_bytes(tiff[off:off + 2], endian)
|
||||
|
||||
def u32(off: int) -> int:
|
||||
return int.from_bytes(tiff[off:off + 4], endian)
|
||||
|
||||
def rational(off: int) -> float:
|
||||
num = u32(off)
|
||||
den = u32(off + 4)
|
||||
return num / den if den else 0.0
|
||||
|
||||
def ascii_value(value: int, count: int, entry_off: int) -> str:
|
||||
raw = tiff[entry_off + 8:entry_off + 8 + count] if count <= 4 else tiff[value:value + count]
|
||||
return raw.split(b"\x00", 1)[0].decode("ascii", "ignore").strip()
|
||||
|
||||
def rationals(value: int, count: int) -> list[float]:
|
||||
return [rational(value + i * 8) for i in range(count)]
|
||||
|
||||
try:
|
||||
if u16(2) != 42:
|
||||
return None
|
||||
ifd0 = u32(4)
|
||||
entries = u16(ifd0)
|
||||
gps_ifd = None
|
||||
for i in range(entries):
|
||||
off = ifd0 + 2 + i * 12
|
||||
if off + 12 > len(tiff):
|
||||
break
|
||||
if u16(off) == 0x8825:
|
||||
gps_ifd = u32(off + 8)
|
||||
break
|
||||
if gps_ifd is None or gps_ifd + 2 > len(tiff):
|
||||
return None
|
||||
gps_entries = u16(gps_ifd)
|
||||
lat_ref = lon_ref = ""
|
||||
lat = lon = None
|
||||
for i in range(gps_entries):
|
||||
off = gps_ifd + 2 + i * 12
|
||||
if off + 12 > len(tiff):
|
||||
break
|
||||
tag = u16(off)
|
||||
typ = u16(off + 2)
|
||||
count = u32(off + 4)
|
||||
value = u32(off + 8)
|
||||
if tag == 1 and typ == 2:
|
||||
lat_ref = ascii_value(value, count, off)
|
||||
elif tag == 2 and typ == 5:
|
||||
lat = rationals(value, count)
|
||||
elif tag == 3 and typ == 2:
|
||||
lon_ref = ascii_value(value, count, off)
|
||||
elif tag == 4 and typ == 5:
|
||||
lon = rationals(value, count)
|
||||
if not lat or not lon or len(lat) < 3 or len(lon) < 3:
|
||||
return None
|
||||
lat_dec = lat[0] + lat[1] / 60 + lat[2] / 3600
|
||||
lon_dec = lon[0] + lon[1] / 60 + lon[2] / 3600
|
||||
if lat_ref.upper() == "S":
|
||||
lat_dec = -lat_dec
|
||||
if lon_ref.upper() == "W":
|
||||
lon_dec = -lon_dec
|
||||
return lat_dec, lon_dec
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def ffprobe_gps(path: Path) -> tuple[float, float] | None:
|
||||
try:
|
||||
out = capture([
|
||||
tool_path(FFPROBE, "ffprobe"), "-v", "error",
|
||||
"-show_format", "-show_streams",
|
||||
"-of", "json",
|
||||
str(path),
|
||||
])
|
||||
data = json.loads(out)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
values: list[str] = []
|
||||
def collect(obj: object) -> None:
|
||||
if isinstance(obj, dict):
|
||||
tags = obj.get("tags")
|
||||
if isinstance(tags, dict):
|
||||
for key, value in tags.items():
|
||||
if "location" in key.lower() or "gps" in key.lower():
|
||||
values.append(str(value))
|
||||
for value in obj.values():
|
||||
collect(value)
|
||||
elif isinstance(obj, list):
|
||||
for value in obj:
|
||||
collect(value)
|
||||
collect(data)
|
||||
for value in values:
|
||||
match = re.search(r"([+-]\d+(?:\.\d+)?)([+-]\d+(?:\.\d+)?)", value)
|
||||
if match:
|
||||
return float(match.group(1)), float(match.group(2))
|
||||
return None
|
||||
|
||||
|
||||
def media_gps(path: Path) -> tuple[float, float] | None:
|
||||
if path.suffix.lower() in IMAGE_EXTS:
|
||||
gps = exif_image_gps(path)
|
||||
if gps:
|
||||
return gps
|
||||
return ffprobe_gps(path)
|
||||
|
||||
|
||||
def reverse_geocode_city_country(lat: float, lon: float) -> str | None:
|
||||
last_country = ""
|
||||
for zoom in range(18, 2, -1):
|
||||
query = urllib.parse.urlencode({
|
||||
"format": "jsonv2",
|
||||
"lat": f"{lat:.8f}",
|
||||
"lon": f"{lon:.8f}",
|
||||
"zoom": str(zoom),
|
||||
"addressdetails": "1",
|
||||
})
|
||||
request = urllib.request.Request(
|
||||
f"https://nominatim.openstreetmap.org/reverse?{query}",
|
||||
headers={"User-Agent": "movmaker/1.0 (reverse geocoding for local movie metadata)"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=10) as response:
|
||||
data = json.loads(response.read().decode("utf-8"))
|
||||
address = data.get("address", {})
|
||||
country = address.get("country", "")
|
||||
last_country = country or last_country
|
||||
city = next((address.get(key) for key in ("city", "town", "village") if address.get(key)), "")
|
||||
if city:
|
||||
return ", ".join(x for x in [city, country] if x)
|
||||
except Exception:
|
||||
continue
|
||||
return last_country or None
|
||||
|
||||
|
||||
def infer_place_from_gps(media: list[MediaItem]) -> str | None:
|
||||
for item in media:
|
||||
gps = media_gps(item.path)
|
||||
if not gps:
|
||||
continue
|
||||
place = reverse_geocode_city_country(*gps)
|
||||
if place:
|
||||
print(f"Location from GPS: {place} ({gps[0]:.6f}, {gps[1]:.6f})")
|
||||
return place
|
||||
return None
|
||||
|
||||
|
||||
def media_date(path: Path) -> datetime:
|
||||
ext = path.suffix.lower()
|
||||
dt = None
|
||||
|
|
@ -441,8 +602,8 @@ def build_video(
|
|||
final_label = "vout"
|
||||
title_label = "titled"
|
||||
title_font = str(Path.home() / ".local/share/fonts/google/BebasNeue-Regular.ttf")
|
||||
stamp_font = str(Path.home() / ".local/share/fonts/google/Inter%5Bopsz,wght%5D.ttf")
|
||||
stamp_bold_font = str(Path.home() / ".local/share/fonts/google/Inter%5Bopsz,wght%5D.ttf")
|
||||
stamp_font = str(Path.home() / ".local/share/fonts/google/NotoSans-SemiBold.ttf")
|
||||
stamp_bold_font = str(Path.home() / ".local/share/fonts/google/NotoSans-SemiBold.ttf")
|
||||
if title_lines:
|
||||
title_font_size = max(60, height // 10)
|
||||
subtitle_font_size = max(40, height // 18)
|
||||
|
|
@ -600,6 +761,8 @@ def main(argv: list[str] | None = None) -> int:
|
|||
first_media_date = media_date(media[0].path)
|
||||
if not data.date:
|
||||
data.date = display_date(first_media_date)
|
||||
if not data.place:
|
||||
data.place = infer_place_from_gps(media) or ""
|
||||
fade = min(args.fade, max(0.0, args.image_duration - 0.1))
|
||||
audio_fade = max(0.0, args.audio_fade)
|
||||
audio_fade_in = max(0.0, args.audio_fade_in)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue