location: add all.py, using takeout/gpslogger/ip (#237)

* location: add all.py, using takeout/gpslogger/ip, update docs

my/location/all.py

@@ -0,0 +1,48 @@
+"""
+Merges location data from multiple sources
+"""
+
+from typing import Iterator
+
+from my.core import Stats, LazyLogger
+from my.core.source import import_source
+
+from my.location.via_ip import locations
+
+from .common import Location
+
+
+logger = LazyLogger(__name__, level="warning")
+
+
+def locations() -> Iterator[Location]:
+    # can add/comment out sources here to disable them, or use core.disabled_modules
+    yield from _takeout_locations()
+    yield from _gpslogger_locations()
+    yield from _ip_locations()
+
+
+@import_source(module_name="my.location.google_takeout")
+def _takeout_locations() -> Iterator[Location]:
+    from . import google_takeout
+    yield from google_takeout.locations()
+
+
+@import_source(module_name="my.location.gpslogger")
+def _gpslogger_locations() -> Iterator[Location]:
+    from . import gpslogger
+    yield from gpslogger.locations()
+
+
+@import_source(module_name="my.location.via_ip")
+def _ip_locations() -> Iterator[Location]:
+    from . import via_ip
+    yield from via_ip.locations()
+
+
+def stats() -> Stats:
+    from my.core import stat
+
+    return {
+        **stat(locations),
+    }

my/location/common.py

@@ -0,0 +1,17 @@
+from datetime import date, datetime
+from typing import Union, Tuple, NamedTuple, Optional
+
+from my.core import __NOT_HPI_MODULE__
+
+DateIsh = Union[datetime, date, str]
+
+LatLon = Tuple[float, float]
+
+
+# TODO: add timezone to this? can use timezonefinder in tz provider instead though
+class Location(NamedTuple):
+    lat: float
+    lon: float
+    dt: datetime
+    accuracy: Optional[float]
+    elevation: Optional[float]

my/location/google.py

@@ -1,6 +1,9 @@
 """
 Location data from Google Takeout
+
+DEPRECATED: setup my.google.takeout.parser and use my.location.google_takeout instead
 """
+
 REQUIRES = [
     'geopy', # checking that coordinates are valid
     'ijson',
@@ -20,6 +23,10 @@ from ..core.common import LazyLogger, mcachew
 from ..core.cachew import cache_dir
 from ..core import kompress
 
+from my.core.warnings import high
+
+high("Please set up my.google.takeout.parser module for better takeout support")
+
 
  # otherwise uses ijson
  # todo move to config??

my/location/google_takeout.py

@@ -0,0 +1,33 @@
+"""
+Extracts locations using google_takeout_parser -- no shared code with the deprecated my.location.google
+"""
+
+REQUIRES = ["git+https://github.com/seanbreckenridge/google_takeout_parser"]
+
+from typing import Iterator
+
+from my.google.takeout.parser import events, _cachew_depends_on
+from google_takeout_parser.models import Location as GoogleLocation
+
+from my.core.common import mcachew, LazyLogger, Stats
+from .common import Location
+
+logger = LazyLogger(__name__)
+
+
+@mcachew(
+    depends_on=_cachew_depends_on,
+    logger=logger,
+)
+def locations() -> Iterator[Location]:
+    for g in events():
+        if isinstance(g, GoogleLocation):
+            yield Location(
+                lon=g.lng, lat=g.lat, dt=g.dt, accuracy=g.accuracy, elevation=None
+            )
+
+
+def stats() -> Stats:
+    from my.core import stat
+
+    return {**stat(locations)}

my/location/gpslogger.py

@@ -0,0 +1,74 @@
+"""
+Parse [[https://github.com/mendhak/gpslogger][gpslogger]] .gpx (xml) files
+"""
+
+REQUIRES = ["gpxpy"]
+
+from my.config import location
+from my.core import Paths, dataclass
+
+
+@dataclass
+class config(location.gpslogger):
+    # path[s]/glob to the synced gpx (XML) files
+    export_path: Paths
+
+    # default accuracy for gpslogger
+    accuracy: float = 50.0
+
+
+from itertools import chain
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterator, Sequence, List
+
+import gpxpy  # type: ignore[import]
+from more_itertools import unique_everseen
+
+from my.core import Stats, LazyLogger
+from my.core.common import get_files, mcachew
+from .common import Location
+
+
+logger = LazyLogger(__name__, level="warning")
+
+
+def inputs() -> Sequence[Path]:
+    return get_files(config.export_path, glob="*.gpx")
+
+
+def _cachew_depends_on() -> List[float]:
+    return [p.stat().st_mtime for p in inputs()]
+
+
+# TODO: could use a better cachew key/this has to recompute every file whenever the newest one changes
+@mcachew(depends_on=_cachew_depends_on, logger=logger)
+def locations() -> Iterator[Location]:
+    yield from unique_everseen(
+        chain(*map(_extract_locations, inputs())), key=lambda loc: loc.dt
+    )
+
+
+def _extract_locations(path: Path) -> Iterator[Location]:
+    with path.open("r") as gf:
+        gpx_obj = gpxpy.parse(gf)
+        for track in gpx_obj.tracks:
+            for segment in track.segments:
+                for point in segment.points:
+                    if point.time is None:
+                        continue
+                    # hmm - for gpslogger, seems that timezone is always SimpleTZ('Z'), which
+                    # specifies UTC -- see https://github.com/tkrajina/gpxpy/blob/cb243b22841bd2ce9e603fe3a96672fc75edecf2/gpxpy/gpxfield.py#L38
+                    yield Location(
+                        lat=point.latitude,
+                        lon=point.longitude,
+                        accuracy=config.accuracy,
+                        elevation=point.elevation,
+                        dt=datetime.replace(point.time, tzinfo=timezone.utc),
+                    )
+
+
+def stats() -> Stats:
+    from my.core import stat
+
+    return {**stat(locations)}

my/location/home.py

@@ -2,17 +2,13 @@
 Simple location provider, serving as a fallback when more detailed data isn't available
 '''
 from dataclasses import dataclass
-from datetime import datetime, date, time, timezone
+from datetime import datetime, time, timezone
 from functools import lru_cache
 from typing import Sequence, Tuple, Union, cast
 
 from my.config import location as user_config
 
-
-DateIsh = Union[datetime, date, str]
-
-# todo hopefully reasonable? might be nice to add name or something too
-LatLon = Tuple[float, float]
+from my.location.common import LatLon, DateIsh
 
 @dataclass
 class Config(user_config):

my/location/via_ip.py

@@ -0,0 +1,39 @@
+"""
+Converts IP addresses provided by my.location.ip to estimated locations
+"""
+
+REQUIRES = ["git+https://github.com/seanbreckenridge/ipgeocache"]
+
+from my.core import dataclass, Stats
+from my.config import location
+
+
+@dataclass
+class config(location.via_ip):
+    # no real science to this, just a guess of ~15km accuracy for IP addresses
+    accuracy: float = 15_000.0
+
+
+from typing import Iterator
+
+from .common import Location
+from my.ip.all import ips
+
+
+def locations() -> Iterator[Location]:
+    for ip in ips():
+        loc: str = ip.ipgeocache()["loc"]
+        lat, _, lon = loc.partition(",")
+        yield Location(
+            lat=float(lat),
+            lon=float(lon),
+            dt=ip.dt,
+            accuracy=config.accuracy,
+            elevation=None,
+        )
+
+
+def stats() -> Stats:
+    from my.core import stat
+
+    return {**stat(locations)}