diff --git a/my/location/all.py b/my/location/all.py index eec4bcc..8d51a82 100644 --- a/my/location/all.py +++ b/my/location/all.py @@ -32,6 +32,7 @@ def _gpslogger_locations() -> Iterator[Location]: yield from gpslogger.locations() +# TODO: remove, user should use fallback.estimate_location or fallback.fallback_locations instead @import_source(module_name="my.location.via_ip") def _ip_locations() -> Iterator[Location]: from . import via_ip diff --git a/my/location/fallback/all.py b/my/location/fallback/all.py index acfa9d5..0c7b8cd 100644 --- a/my/location/fallback/all.py +++ b/my/location/fallback/all.py @@ -1,7 +1,7 @@ # TODO: add config here which passes kwargs to estimate_from (under_accuracy) # overwritable by passing the kwarg name here to the top-level estimate_location -from typing import Iterator +from typing import Iterator, Optional from my.core.source import import_source from my.location.fallback.common import ( @@ -12,8 +12,8 @@ from my.location.fallback.common import ( ) -# can comment/uncomment sources here to enable/disable them def fallback_locations() -> Iterator[FallbackLocation]: + # can comment/uncomment sources here to enable/disable them yield from _ip_fallback_locations() @@ -24,8 +24,8 @@ def fallback_estimators() -> Iterator[LocationEstimator]: yield _home_estimate -def estimate_location(dt: DateExact) -> FallbackLocation: - loc = estimate_from(dt, estimators=list(fallback_estimators())) +def estimate_location(dt: DateExact, first_match: bool=False, under_accuracy: Optional[int] = None) -> FallbackLocation: + loc = estimate_from(dt, estimators=list(fallback_estimators()), first_match=first_match, under_accuracy=under_accuracy) # should never happen if the user has home configured if loc is None: raise ValueError("Could not estimate location") diff --git a/my/location/fallback/via_ip.py b/my/location/fallback/via_ip.py index 98197c2..e4c35a8 100644 --- a/my/location/fallback/via_ip.py +++ b/my/location/fallback/via_ip.py @@ -86,8 +86,9 @@ def estimate_location(dt: DateExact) -> Iterator[FallbackLocation]: if start_time <= dt_ts <= end_time: # logger.debug(f"Found location for {dt}: {loc}") yield loc - if end_time > dt_ts: - # logger.debug(f"Passed end time: {end_time} > {dt_ts} ({datetime.fromtimestamp(end_time)} > {datetime.fromtimestamp(dt_ts)})") + # no more locations could possibly contain dt + if start_time > dt_ts: + # logger.debug(f"Passed start time: {end_time} > {dt_ts} ({datetime.fromtimestamp(end_time)} > {datetime.fromtimestamp(dt_ts)})") break idx += 1 diff --git a/tests/location.py b/tests/location.py index 298b7ba..c47849e 100644 --- a/tests/location.py +++ b/tests/location.py @@ -1,7 +1,5 @@ from pathlib import Path -from more_itertools import one - import pytest # type: ignore @@ -20,26 +18,11 @@ def test() -> None: @pytest.fixture(autouse=True) def prepare(tmp_path: Path): - from .common import reset_modules - reset_modules() - - user_config = _prepare_google_config(tmp_path) + from .shared_config import temp_config + user_config = temp_config(tmp_path) import my.core.cfg as C with C.tmp_config() as config: - config.google = user_config # type: ignore + config.google = user_config.google yield - -def _prepare_google_config(tmp_path: Path): - from .common import testdata - track = one(testdata().rglob('italy-slovenia-2017-07-29.json')) - - # todo ugh. unnecessary zipping, but at the moment takeout provider doesn't support plain dirs - import zipfile - with zipfile.ZipFile(tmp_path / 'takeout.zip', 'w') as zf: - zf.writestr('Takeout/Location History/Location History.json', track.read_bytes()) - - class google_config: - takeout_path = tmp_path - return google_config diff --git a/tests/location_fallback.py b/tests/location_fallback.py new file mode 100644 index 0000000..0d291c2 --- /dev/null +++ b/tests/location_fallback.py @@ -0,0 +1,124 @@ +""" +To test my.location.fallback_location.all +""" + +from typing import Iterator +from datetime import datetime, timezone, timedelta + +from more_itertools import ilen + +from my.ip.common import IP + +def data() -> Iterator[IP]: + # random IP addresses + yield IP(addr="67.98.113.0", dt=datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc)) + yield IP(addr="67.98.112.0", dt=datetime(2020, 1, 15, 12, 0, 0, tzinfo=timezone.utc)) + yield IP(addr="59.40.113.87", dt=datetime(2020, 2, 1, 12, 0, 0, tzinfo=timezone.utc)) + yield IP(addr="59.40.139.87", dt=datetime(2020, 2, 1, 16, 0, 0, tzinfo=timezone.utc)) + yield IP(addr="161.235.192.228", dt=datetime(2020, 3, 1, 12, 0, 0, tzinfo=timezone.utc)) + +# redefine the my.ip.all function using data for testing +import my.ip.all as ip_module +ip_module.ips = data + +from my.location.fallback import via_ip + +# these are all tests for the bisect algorithm defined in via_ip.py +# to make sure we can correctly find IPs that are within the 'for_duration' of a given datetime + +def test_ip_fallback() -> None: + # make sure that the data override works + assert ilen(ip_module.ips()) == ilen(data()) + assert ilen(ip_module.ips()) == ilen(via_ip.fallback_locations()) + assert ilen(via_ip.fallback_locations()) == 5 + assert ilen(via_ip._sorted_fallback_locations()) == 5 + + # confirm duration from via_ip since that is used for bisect + assert via_ip.config.for_duration == timedelta(hours=24) + + # basic tests + + # try estimating slightlight before the first IP + est = list(via_ip.estimate_location(datetime(2020, 1, 1, 11, 59, 59, tzinfo=timezone.utc))) + assert len(est) == 0 + + # during the duration for the first IP + est = list(via_ip.estimate_location(datetime(2020, 1, 1, 12, 30, 0, tzinfo=timezone.utc))) + assert len(est) == 1 + + # right after the 'for_duration' for an IP + est = list(via_ip.estimate_location(datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + via_ip.config.for_duration + timedelta(seconds=1))) + assert len(est) == 0 + + # on 2/1/2020, theres one IP if before 16:30 + est = list(via_ip.estimate_location(datetime(2020, 2, 1, 12, 30, 0, tzinfo=timezone.utc))) + assert len(est) == 1 + + # and two if after 16:30 + est = list(via_ip.estimate_location(datetime(2020, 2, 1, 17, 00, 0, tzinfo=timezone.utc))) + assert len(est) == 2 + + # the 12:30 IP should 'expire' before the 16:30 IP, use 3:30PM on the next day + est = list(via_ip.estimate_location(datetime(2020, 2, 2, 15, 30, 0, tzinfo=timezone.utc))) + assert len(est) == 1 + + use_dt = datetime(2020, 3, 1, 12, 15, 0, tzinfo=timezone.utc) + + # test last IP + est = list(via_ip.estimate_location(use_dt)) + assert len(est) == 1 + + # datetime should be the IPs, not the passed IP (if via_home, it uses the passed dt) + assert est[0].dt != use_dt + + # test interop with other fallback estimators/all.py + # + # redefine fallback_estimators to prevent possible namespace packages the user + # may have installed from having side effects testing this + from my.location.fallback import all + from my.location.fallback import via_home + def _fe() -> Iterator[all.LocationEstimator]: + yield via_ip.estimate_location + yield via_home.estimate_location + + all.fallback_estimators = _fe + assert ilen(all.fallback_estimators()) == 2 + + # test that all.estimate_location has access to both IPs + # + # just passing via_ip should give one IP + from my.location.fallback.common import _iter_estimate_from + raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location,))) + assert len(raw_est) == 1 + assert raw_est[0].accuracy == 10_000 + + # passing home should give one + home_est = list(_iter_estimate_from(use_dt, (via_home.estimate_location,))) + assert len(home_est) == 1 + assert home_est[0].accuracy == 30_000 + + # make sure ip accuracy is more accurate + assert raw_est[0].accuracy < home_est[0].accuracy + + # passing both should give two + raw_est = list(_iter_estimate_from(use_dt, (via_ip.estimate_location, via_home.estimate_location))) + assert len(raw_est) == 2 + + # shouldnt raise value error + all_est = all.estimate_location(use_dt) + # should have used the IP from via_ip since it was more accurate + assert all_est.datasource == "via_ip" + + # test that a home defined in shared_config.py is used if no IP is found + loc = all.estimate_location(datetime(2021, 1, 1, 12, 30, 0, tzinfo=timezone.utc)) + assert loc.datasource == "via_home" + + # test a different home using location.fallback.all + bulgaria = all.estimate_location(datetime(2006, 1, 1, 12, 30, 0, tzinfo=timezone.utc)) + assert bulgaria.datasource == "via_home" + assert (bulgaria.lat, bulgaria.lon) == (42.697842, 23.325973) + assert (loc.lat, loc.lon) != (bulgaria.lat, bulgaria.lon) + + +# re-use prepare fixture for overriding config from shared_config.py +from .tz import prepare diff --git a/tests/shared_config.py b/tests/shared_config.py new file mode 100644 index 0000000..c8eec50 --- /dev/null +++ b/tests/shared_config.py @@ -0,0 +1,64 @@ +# Defines some shared config for tests + +from datetime import datetime, date, timezone +from pathlib import Path + +from typing import Any, NamedTuple +import my.time.tz.via_location as LTZ +from more_itertools import one + + +class SharedConfig(NamedTuple): + google: Any + location: Any + time: Any + + +def _prepare_google_config(tmp_path: Path): + from .common import testdata + try: + track = one(testdata().rglob('italy-slovenia-2017-07-29.json')) + except ValueError: + raise RuntimeError('testdata not found, setup git submodules?') + + + # todo ugh. unnecessary zipping, but at the moment takeout provider doesn't support plain dirs + import zipfile + with zipfile.ZipFile(tmp_path / 'takeout.zip', 'w') as zf: + zf.writestr('Takeout/Location History/Location History.json', track.read_bytes()) + + class google_config: + takeout_path = tmp_path + return google_config + + +# pass tmp_path from pytest to this helper function +# see tests/tz.py as an example +def temp_config(temp_path: Path) -> Any: + from .common import reset_modules + reset_modules() + + LTZ.config.fast = True + + class location: + home = ( + # supports ISO strings + ('2005-12-04' , (42.697842, 23.325973)), # Bulgaria, Sofia + # supports date/datetime objects + (date(year=1980, month=2, day=15) , (40.7128 , -74.0060 )), # NY + # check tz handling.. + (datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558 , 37.6173 )), # Moscow, Russia + ) + # note: order doesn't matter, will be sorted in the data provider + class via_ip: + pass + class gpslogger: + pass + + class time: + class tz: + class via_location: + pass # just rely on the defaults... + + + return SharedConfig(google=_prepare_google_config(temp_path), location=location, time=time) diff --git a/tests/tz.py b/tests/tz.py index a36f75d..8f80800 100644 --- a/tests/tz.py +++ b/tests/tz.py @@ -1,5 +1,5 @@ import sys -from datetime import datetime, timedelta, date, timezone +from datetime import datetime, timedelta from pathlib import Path import pytest # type: ignore @@ -81,40 +81,15 @@ def D(dstr: str) -> datetime: return datetime.strptime(dstr, '%Y%m%d %H:%M:%S') -# TODO copy pasted from location.py, need to extract some common provider + @pytest.fixture(autouse=True) def prepare(tmp_path: Path): - from .common import reset_modules - reset_modules() - - LTZ.config.fast = True - - from .location import _prepare_google_config - google = _prepare_google_config(tmp_path) - - class location: - home = ( - # supports ISO strings - ('2005-12-04' , (42.697842, 23.325973)), # Bulgaria, Sofia - # supports date/datetime objects - (date(year=1980, month=2, day=15) , (40.7128 , -74.0060 )), # NY - # check tz handling.. - (datetime.fromtimestamp(1600000000, tz=timezone.utc), (55.7558 , 37.6173 )), # Moscow, Russia - ) - # note: order doesn't matter, will be sorted in the data provider - class via_ip: - pass - class gpslogger: - pass - - class time: - class tz: - class via_location: - pass # just rely on the defaults... + from .shared_config import temp_config + conf = temp_config(tmp_path) import my.core.cfg as C with C.tmp_config() as config: - config.google = google - config.time = time - config.location = location + config.google = conf.google + config.time = conf.time + config.location = conf.location yield