From bbd56d23b4b64f88071ac853f1f56cba5d2e47bf Mon Sep 17 00:00:00 2001 From: marc7s Date: Sun, 3 Nov 2024 20:11:52 +0100 Subject: [PATCH 01/37] Fix missing async keyword, see https://github.com/home-assistant/core/issues/99977 --- geocachingapi/geocachingapi.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 036f1ac..94c0298 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -90,7 +90,7 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: self._close_session = True try: - with async_timeout.timeout(self.request_timeout): + async with async_timeout.timeout(self.request_timeout): response = await self._session.request( method, f"{url}", From a66e14e61152b021334558d2acd4af9986fa245f Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 4 Nov 2024 12:16:18 +0100 Subject: [PATCH 02/37] Add nearby caches to status --- geocachingapi/geocachingapi.py | 26 +++++++++++++++++++- geocachingapi/models.py | 43 +++++++++++++++++++++++++++++++--- 2 files changed, 65 insertions(+), 4 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 94c0298..39f9b05 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -21,6 +21,7 @@ ) from .models import ( + GeocachingCoordinate, GeocachingStatus, GeocachingSettings, GeocachingApiEnvironment, @@ -137,9 +138,11 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: return result async def update(self) -> GeocachingStatus: - await self._update_user(None) + await self._update_user() if len(self._settings.trackable_codes) > 0: await self._update_trackables() + if self._settings.nearby_caches_setting is not None: + await self._update_nearby_caches() _LOGGER.info(f'Status updated.') return self._status @@ -188,6 +191,27 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: _LOGGER.debug(f'Trackables updated.') + async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: + assert self._status + if self._settings.nearby_caches_setting is None: + _LOGGER.warning("Cannot update nearby caches, setting has not been configured.") + return + + if data is None: + fields = ",".join([ + "referenceCode", + "name", + "postedCoordinates" + ]) + coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location + radiusKm: float = self._settings.nearby_caches_setting.radiusKm + URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusKm}km&fields={fields}&sort=distance+&lite=true" + # The + sign is not encoded correctly, so we encode it manually + data = await self._request("GET", URL.replace("+", "%2B")) + self._status.update_nearby_caches_from_dict(data) + + _LOGGER.debug(f'Nearby caches updated.') + async def update_settings(self, settings: GeocachingSettings): """Update the Geocaching settings""" self._settings = settings diff --git a/geocachingapi/models.py b/geocachingapi/models.py index de46266..e856c65 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -19,17 +19,27 @@ class GeocachingApiEnvironment(Enum): Staging = 1, Production = 2, +@dataclass +class NearbyCachesSetting: + location: GeocachingCoordinate + radiusKm: float + class GeocachingSettings: """Class to hold the Geocaching Api settings""" trackable_codes: array(str) environment: GeocachingApiEnvironment + nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables:array(str) = [] ) -> None: + def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: array(str) = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.trackable_codes = trackables + self.nearby_caches_setting = nearby_caches_setting - def set_trackables(self, trackables:array(str)): + def set_trackables(self, trackables: array(str)): self.trackable_codes = trackables + + def set_nearby_caches_setting(self, setting: NearbyCachesSetting): + self.nearby_caches_setting = setting @dataclass class GeocachingUser: @@ -137,12 +147,26 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.trackable_type = try_get_from_dict(data, "type", self.trackable_type) if "trackableLogs" in data and len(data["trackableLogs"]) > 0: self.latest_log = GeocachingTrackableLog(data=data["trackableLogs"][0]) - + +@dataclass +class GeocachingCache: + reference_code: Optional[str] = None + name: Optional[str] = None + coordinates: GeocachingCoordinate = None + + def update_from_dict(self, data: Dict[str, Any]) -> None: + self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) + self.name = try_get_from_dict(data, "name", self.name) + if "coordinates" in data: + self.coordinates = GeocachingCoordinate(data=data["coordinates"]) + else: + self.coordinates = None class GeocachingStatus: """Class to hold all account status information""" user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None + nearby_caches: list[GeocachingCache] = None def __init__(self): """Initialize GeocachingStatus""" @@ -162,4 +186,17 @@ def update_trackables_from_dict(self, data: Any) -> None: if not reference_code in self.trackables.keys(): self.trackables[reference_code] = GeocachingTrackable() self.trackables[reference_code].update_from_dict(trackable) + + def update_nearby_caches_from_dict(self, data: Any) -> None: + """Update nearby caches from the API result""" + if not any(data): + pass + + nearby_caches: list[GeocachingCache] = [] + for cacheData in data: + cache = GeocachingCache() + cache.update_from_dict(cacheData) + nearby_caches.append(cache) + + self.nearby_caches = nearby_caches \ No newline at end of file From 5f1a5d88f3cc7230e8069a826f21a224784f0278 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 4 Nov 2024 14:13:19 +0100 Subject: [PATCH 03/37] Change trackable distance traveled data type --- geocachingapi/models.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index e856c65..e79ad44 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -117,8 +117,8 @@ class GeocachingTrackable: name: Optional[str] = None holder: GeocachingUser = None tracking_number: Optional[str] = None - kilometers_traveled: Optional[str] = None - miles_traveled: Optional[str] = None + kilometers_traveled: Optional[float] = None + miles_traveled: Optional[float] = None current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None latest_journey: GeocachingTrackableJourney = None @@ -139,8 +139,8 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: holder = None self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) - self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled) - self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled) + self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) + self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled, float) self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code) self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name) self.is_missing = try_get_from_dict(data, "isMissing", self.is_missing) From a8208114d4e8198cb17486611a3e7fe400c4a528 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 20 Nov 2024 16:21:41 +0100 Subject: [PATCH 04/37] Fix missing status initialization for nearby caches --- geocachingapi/models.py | 1 + 1 file changed, 1 insertion(+) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index e79ad44..7d5d4c9 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -172,6 +172,7 @@ def __init__(self): """Initialize GeocachingStatus""" self.user = GeocachingUser() self.trackables = {} + self.nearby_caches = [] def update_user_from_dict(self, data: Dict[str, Any]) -> None: """Update user from the API result""" From f0631a10ec3daccd4529b45b7732103221779851 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 21 Nov 2024 18:42:31 +0100 Subject: [PATCH 05/37] Fix typo causing cache coordinates not to be parsed --- geocachingapi/models.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 7d5d4c9..7bf1dd1 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -157,8 +157,8 @@ class GeocachingCache: def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) - if "coordinates" in data: - self.coordinates = GeocachingCoordinate(data=data["coordinates"]) + if "postedCoordinates" in data: + self.coordinates = GeocachingCoordinate(data=data["postedCoordinates"]) else: self.coordinates = None From a018bed28b05679a377e9ff55cc6a1de742b2bd6 Mon Sep 17 00:00:00 2001 From: Per Samuelsson Date: Fri, 22 Nov 2024 18:05:44 +0100 Subject: [PATCH 06/37] Added update_caches and _get_cache_info, with some more stuff, have not been able to test that it works yet, /Per&Albin --- geocachingapi/geocachingapi.py | 17 +++- geocachingapi/models.py | 137 +++++++++++++++++++++++++-------- 2 files changed, 119 insertions(+), 35 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 39f9b05..01f190f 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -145,7 +145,22 @@ async def update(self) -> GeocachingStatus: await self._update_nearby_caches() _LOGGER.info(f'Status updated.') return self._status - + + async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: + assert self._status + if data is None: + fields = ",".join([ + "refrenceCode", + "name", + "postedCoordinates", + "favoritePoints" + ]) + caches_parameters = ",".join(self._settings.caches_codes) + # Need to send reference code of caches that we want + data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}lite=true&fields={fields}" ) + self._status.update_caches(data) + _LOGGER.debug(f'Caches updated.') + async def _update_user(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 7bf1dd1..eace8a0 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -7,43 +7,63 @@ from datetime import datetime from .utils import try_get_from_dict + class GeocachingApiEnvironmentSettings(TypedDict): """Class to represent API environment settings""" - api_scheme:str - api_host:str + + api_scheme: str + api_host: str api_port: int - api_base_bath:str + api_base_bath: str + class GeocachingApiEnvironment(Enum): """Enum to represent API environment""" - Staging = 1, - Production = 2, + + Staging = (1,) + Production = (2,) + @dataclass class NearbyCachesSetting: location: GeocachingCoordinate radiusKm: float + class GeocachingSettings: """Class to hold the Geocaching Api settings""" + + caches_codes: array(str) trackable_codes: array(str) environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: array(str) = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__( + self, + environment: GeocachingApiEnvironment = GeocachingApiEnvironment.Production, + trackables: array(str) = [], + caches: array(str) = [], + nearby_caches_setting: NearbyCachesSetting = None, + ) -> None: """Initialize settings""" self.trackable_codes = trackables self.nearby_caches_setting = nearby_caches_setting - + self.caches_codes = caches + + def set_caches(self, caches: array(str)): + self.caches_codes = caches + def set_trackables(self, trackables: array(str)): self.trackable_codes = trackables - + def set_nearby_caches_setting(self, setting: NearbyCachesSetting): self.nearby_caches_setting = setting + @dataclass class GeocachingUser: """Class to hold the Geocaching user information""" + reference_code: Optional[str] = None username: Optional[str] = None find_count: Optional[int] = None @@ -55,18 +75,30 @@ class GeocachingUser: def update_from_dict(self, data: Dict[str, Any]) -> None: """Update user from the API result""" - self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) + self.reference_code = try_get_from_dict( + data, "referenceCode", self.reference_code + ) self.username = try_get_from_dict(data, "username", self.username) self.find_count = try_get_from_dict(data, "findCount", self.find_count) self.hide_count = try_get_from_dict(data, "hideCount", self.hide_count) - self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points) - self.souvenir_count = try_get_from_dict(data, "souvenirCount", self.souvenir_count) - self.awarded_favorite_points = try_get_from_dict(data, "awardedFavoritePoints", self.awarded_favorite_points) - self.membership_level_id = try_get_from_dict(data, "membershipLevelId", self.membership_level_id) + self.favorite_points = try_get_from_dict( + data, "favoritePoints", self.favorite_points + ) + self.souvenir_count = try_get_from_dict( + data, "souvenirCount", self.souvenir_count + ) + self.awarded_favorite_points = try_get_from_dict( + data, "awardedFavoritePoints", self.awarded_favorite_points + ) + self.membership_level_id = try_get_from_dict( + data, "membershipLevelId", self.membership_level_id + ) + @dataclass class GeocachingCoordinate: """Class to hold a Geocaching coordinate""" + latitude: Optional[str] = None longitude: Optional[str] = None @@ -75,9 +107,11 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: self.latitude = try_get_from_dict(data, "latitude", None) self.longitude = try_get_from_dict(data, "longitude", None) + @dataclass class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" + coordinates: GeocachingCoordinate = None logged_date: Optional[datetime] = None @@ -89,6 +123,7 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: self.coordinates = None self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) + @dataclass class GeocachingTrackableLog: reference_code: Optional[str] = None @@ -98,21 +133,26 @@ class GeocachingTrackableLog: logged_date: Optional[datetime] = None def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableLog: - self.reference_code = try_get_from_dict(data, 'referenceCode',self.reference_code) + self.reference_code = try_get_from_dict( + data, "referenceCode", self.reference_code + ) if self.owner is None: self.owner = GeocachingUser() - if 'owner' in data: - self.owner.update_from_dict(data['owner']) + if "owner" in data: + self.owner.update_from_dict(data["owner"]) else: self.owner = None - self.log_type = try_get_from_dict(data['trackableLogType'], 'name',self.log_type) - self.logged_date = try_get_from_dict(data, 'loggedDate',self.logged_date) - self.text = try_get_from_dict(data, 'text',self.text) + self.log_type = try_get_from_dict( + data["trackableLogType"], "name", self.log_type + ) + self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) + self.text = try_get_from_dict(data, "text", self.text) @dataclass class GeocachingTrackable: """Class to hold the Geocaching trackable information""" + reference_code: Optional[str] = None name: Optional[str] = None holder: GeocachingUser = None @@ -122,62 +162,92 @@ class GeocachingTrackable: current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None latest_journey: GeocachingTrackableJourney = None - is_missing: bool = False, - trackable_type: str = None, + is_missing: bool = (False,) + trackable_type: str = (None,) latest_log: GeocachingTrackableLog = None - def update_from_dict(self, data: Dict[str, Any]) -> None: """Update trackable from the API""" - self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) + self.reference_code = try_get_from_dict( + data, "referenceCode", self.reference_code + ) self.name = try_get_from_dict(data, "name", self.name) if data["holder"] is not None: - if self.holder is None : + if self.holder is None: holder = GeocachingUser() holder.update_from_dict(data["holder"]) else: holder = None - self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) - self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) - self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled, float) - self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code) - self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name) + self.tracking_number = try_get_from_dict( + data, "trackingNumber", self.tracking_number + ) + self.kilometers_traveled = try_get_from_dict( + data, "kilometersTraveled", self.kilometers_traveled, float + ) + self.miles_traveled = try_get_from_dict( + data, "milesTraveled", self.miles_traveled, float + ) + self.current_geocache_code = try_get_from_dict( + data, "currectGeocacheCode", self.current_geocache_code + ) + self.current_geocache_name = try_get_from_dict( + data, "currentGeocacheName", self.current_geocache_name + ) self.is_missing = try_get_from_dict(data, "isMissing", self.is_missing) self.trackable_type = try_get_from_dict(data, "type", self.trackable_type) if "trackableLogs" in data and len(data["trackableLogs"]) > 0: self.latest_log = GeocachingTrackableLog(data=data["trackableLogs"][0]) + @dataclass class GeocachingCache: reference_code: Optional[str] = None name: Optional[str] = None coordinates: GeocachingCoordinate = None + # Maybe add favoritePoints here def update_from_dict(self, data: Dict[str, Any]) -> None: - self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) + self.reference_code = try_get_from_dict( + data, "referenceCode", self.reference_code + ) self.name = try_get_from_dict(data, "name", self.name) if "postedCoordinates" in data: self.coordinates = GeocachingCoordinate(data=data["postedCoordinates"]) else: self.coordinates = None + class GeocachingStatus: """Class to hold all account status information""" + user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None nearby_caches: list[GeocachingCache] = None + caches: GeocachingCache = {"name": "hej"} def __init__(self): """Initialize GeocachingStatus""" self.user = GeocachingUser() self.trackables = {} self.nearby_caches = [] + self.caches = [] def update_user_from_dict(self, data: Dict[str, Any]) -> None: """Update user from the API result""" self.user.update_from_dict(data) - + + def update_caches(self, data: Any) -> None: + """Update caches from the API result""" + if not any(data): + pass + caches: list[GeocachingCache] = [] + for cacheData in data: + cache = GeocachingCache() + cache.update_from_dict(cacheData) + caches.append(cache) + self.caches = caches + def update_trackables_from_dict(self, data: Any) -> None: """Update trackables from the API result""" if not any(data): @@ -187,7 +257,7 @@ def update_trackables_from_dict(self, data: Any) -> None: if not reference_code in self.trackables.keys(): self.trackables[reference_code] = GeocachingTrackable() self.trackables[reference_code].update_from_dict(trackable) - + def update_nearby_caches_from_dict(self, data: Any) -> None: """Update nearby caches from the API result""" if not any(data): @@ -198,6 +268,5 @@ def update_nearby_caches_from_dict(self, data: Any) -> None: cache = GeocachingCache() cache.update_from_dict(cacheData) nearby_caches.append(cache) - + self.nearby_caches = nearby_caches - \ No newline at end of file From 6b21457078d4bcadfbe5a885467ca80bb9c1986e Mon Sep 17 00:00:00 2001 From: Albin Date: Sun, 24 Nov 2024 11:30:39 +0100 Subject: [PATCH 07/37] add seperate call for geocaches --- geocachingapi/geocachingapi.py | 12 +++++++++--- geocachingapi/models.py | 4 +++- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 01f190f..1f1d7ce 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -81,7 +81,7 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: headers = {} else: headers = dict(headers) - + print(self.token) headers["Authorization"] = f"Bearer {self.token}" _LOGGER.debug(f'With headers:') _LOGGER.debug(f'{str(headers)}') @@ -143,6 +143,10 @@ async def update(self) -> GeocachingStatus: await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() + + if self._settings.caches_codes is not None: + await self._get_cache_info() + _LOGGER.info(f'Status updated.') return self._status @@ -150,14 +154,16 @@ async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: fields = ",".join([ - "refrenceCode", "name", "postedCoordinates", "favoritePoints" ]) caches_parameters = ",".join(self._settings.caches_codes) # Need to send reference code of caches that we want - data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}lite=true&fields={fields}" ) + try: + data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={fields}") + except: + data = [] self._status.update_caches(data) _LOGGER.debug(f'Caches updated.') diff --git a/geocachingapi/models.py b/geocachingapi/models.py index eace8a0..0791e1d 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -205,6 +205,7 @@ class GeocachingCache: reference_code: Optional[str] = None name: Optional[str] = None coordinates: GeocachingCoordinate = None + favoritePoints: Optional[int] = None # Maybe add favoritePoints here def update_from_dict(self, data: Dict[str, Any]) -> None: @@ -212,6 +213,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: data, "referenceCode", self.reference_code ) self.name = try_get_from_dict(data, "name", self.name) + self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints) if "postedCoordinates" in data: self.coordinates = GeocachingCoordinate(data=data["postedCoordinates"]) else: @@ -224,7 +226,7 @@ class GeocachingStatus: user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None nearby_caches: list[GeocachingCache] = None - caches: GeocachingCache = {"name": "hej"} + caches: GeocachingCache = None def __init__(self): """Initialize GeocachingStatus""" From 425be24d5447dce752c53c966e19f687dbdd13d6 Mon Sep 17 00:00:00 2001 From: Per Samuelsson Date: Sun, 24 Nov 2024 12:51:42 +0100 Subject: [PATCH 08/37] Started working on the trackable journey list, not yet tested /Albin Per --- geocachingapi/geocachingapi.py | 23 +++++++++++++++++++++-- geocachingapi/models.py | 2 +- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 1f1d7ce..233e2ff 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -139,11 +139,12 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: async def update(self) -> GeocachingStatus: await self._update_user() + if self._settings.trackable_codes is not None: + await self._update_trackable_journey() if len(self._settings.trackable_codes) > 0: await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() - if self._settings.caches_codes is not None: await self._get_cache_info() @@ -159,7 +160,6 @@ async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: "favoritePoints" ]) caches_parameters = ",".join(self._settings.caches_codes) - # Need to send reference code of caches that we want try: data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={fields}") except: @@ -184,6 +184,25 @@ async def _update_user(self, data: Dict[str, Any] = None) -> None: self._status.update_user_from_dict(data) _LOGGER.debug(f'User updated.') + async def _update_trackable_journey(self, data: Dict[str, Any] = None) -> None: + assert self._status + if data is None: + fields = ",".join([ + "referenceCode", + "name", + "owner" + ]) + trackable_parameters = ",".join(self._settings.trackable_codes) + data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}") + self._status.update_trackables_from_dict(data) + if len(self._status.trackables) > 0: + for trackable in self._status.trackables.values(): + trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=100") + if len(trackable_parameters) >= 1: + trackable = GeocachingTrackableJourney(data=trackable_journey_data) + self.trackable_journey.append(trackable) + + async def _update_trackables(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 0791e1d..4727371 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -162,6 +162,7 @@ class GeocachingTrackable: current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None latest_journey: GeocachingTrackableJourney = None + trackable_journey: Optional[GeocachingTrackableJourney] = None is_missing: bool = (False,) trackable_type: str = (None,) latest_log: GeocachingTrackableLog = None @@ -206,7 +207,6 @@ class GeocachingCache: name: Optional[str] = None coordinates: GeocachingCoordinate = None favoritePoints: Optional[int] = None - # Maybe add favoritePoints here def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict( From caf5bebec44959eab13b8ea960ce18e2b313ff52 Mon Sep 17 00:00:00 2001 From: Albin Date: Sun, 24 Nov 2024 19:09:42 +0100 Subject: [PATCH 09/37] fix for appending to array in trackable journey api --- geocachingapi/geocachingapi.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 233e2ff..6ae629e 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -199,8 +199,8 @@ async def _update_trackable_journey(self, data: Dict[str, Any] = None) -> None: for trackable in self._status.trackables.values(): trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=100") if len(trackable_parameters) >= 1: - trackable = GeocachingTrackableJourney(data=trackable_journey_data) - self.trackable_journey.append(trackable) + abc = GeocachingTrackableJourney(data=trackable_journey_data) + trackable.trackable_journey.append(abc) async def _update_trackables(self, data: Dict[str, Any] = None) -> None: From 9d4af3a041de9d5a402027ce9309c445668cdb80 Mon Sep 17 00:00:00 2001 From: Albin Date: Sun, 24 Nov 2024 21:02:06 +0100 Subject: [PATCH 10/37] add support for trackable objects --- geocachingapi/geocachingapi.py | 26 ++++++++++++++++++-------- geocachingapi/models.py | 14 +++++++++++--- 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 6ae629e..be69651 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -11,7 +11,7 @@ from yarl import URL from aiohttp import ClientResponse, ClientSession, ClientError -from typing import Any, Awaitable, Callable, Dict, List, Optional +from typing import Any, Awaitable, Callable, Dict, Optional from .const import ENVIRONMENT_SETTINGS from .exceptions import ( GeocachingApiConnectionError, @@ -141,8 +141,8 @@ async def update(self) -> GeocachingStatus: await self._update_user() if self._settings.trackable_codes is not None: await self._update_trackable_journey() - if len(self._settings.trackable_codes) > 0: - await self._update_trackables() + # if len(self._settings.trackable_codes) > 0: + # await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() if self._settings.caches_codes is not None: @@ -190,18 +190,28 @@ async def _update_trackable_journey(self, data: Dict[str, Any] = None) -> None: fields = ",".join([ "referenceCode", "name", - "owner" + "holder", + "trackingNumber", + "kilometersTraveled", + "milesTraveled", + "currentGeocacheCode", + "currentGeocacheName", + "isMissing", + "type" ]) trackable_parameters = ",".join(self._settings.trackable_codes) data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}") self._status.update_trackables_from_dict(data) if len(self._status.trackables) > 0: for trackable in self._status.trackables.values(): - trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=100") - if len(trackable_parameters) >= 1: - abc = GeocachingTrackableJourney(data=trackable_journey_data) - trackable.trackable_journey.append(abc) + trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=1") + if trackable_journey_data: # Ensure data exists + # Create a list of GeocachingTrackableJourney instances + journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) + for i, journey in enumerate(journeys): + # Add each journey to the trackable's trackable_journeys list by index + trackable.trackable_journeys.append(journey) async def _update_trackables(self, data: Dict[str, Any] = None) -> None: assert self._status diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 4727371..0ab18e0 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -2,9 +2,11 @@ from array import array from enum import Enum from typing import Any, Dict, Optional, TypedDict - +from typing import List +from dataclasses import field from dataclasses import dataclass from datetime import datetime + from .utils import try_get_from_dict @@ -123,6 +125,10 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: self.coordinates = None self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) + @classmethod + def from_list(cls, data_list: List[Dict[str, Any]]) -> List[GeocachingTrackableJourney]: + """Creates a list of GeocachingTrackableJourney instances from an array of data""" + return [cls(data=data) for data in data_list] @dataclass class GeocachingTrackableLog: @@ -161,8 +167,10 @@ class GeocachingTrackable: miles_traveled: Optional[float] = None current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None - latest_journey: GeocachingTrackableJourney = None - trackable_journey: Optional[GeocachingTrackableJourney] = None + latest_journey: Optional[GeocachingTrackableJourney] = None, + trackable_journeys: Optional[List[GeocachingTrackableJourney]] = field(default_factory=list) + + is_missing: bool = (False,) trackable_type: str = (None,) latest_log: GeocachingTrackableLog = None From dea14338fb52b1cd2e39fcef842f66505927a74e Mon Sep 17 00:00:00 2001 From: Albin Date: Sun, 24 Nov 2024 21:07:56 +0100 Subject: [PATCH 11/37] uncommented code --- geocachingapi/geocachingapi.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index be69651..ccfd130 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -141,8 +141,8 @@ async def update(self) -> GeocachingStatus: await self._update_user() if self._settings.trackable_codes is not None: await self._update_trackable_journey() - # if len(self._settings.trackable_codes) > 0: - # await self._update_trackables() + if len(self._settings.trackable_codes) > 0: + await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() if self._settings.caches_codes is not None: From 051db11a48cc2ef0b2f9570d9834621a47ad43e6 Mon Sep 17 00:00:00 2001 From: marc7s Date: Sun, 24 Nov 2024 22:32:59 +0100 Subject: [PATCH 12/37] Revert formatting --- geocachingapi/geocachingapi.py | 7 +- geocachingapi/models.py | 118 +++++++++------------------------ 2 files changed, 34 insertions(+), 91 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index ccfd130..bfb85d5 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -81,7 +81,7 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: headers = {} else: headers = dict(headers) - print(self.token) + headers["Authorization"] = f"Bearer {self.token}" _LOGGER.debug(f'With headers:') _LOGGER.debug(f'{str(headers)}') @@ -160,10 +160,7 @@ async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: "favoritePoints" ]) caches_parameters = ",".join(self._settings.caches_codes) - try: - data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={fields}") - except: - data = [] + data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={fields}") self._status.update_caches(data) _LOGGER.debug(f'Caches updated.') diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 0ab18e0..ce83979 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -1,52 +1,36 @@ from __future__ import annotations from array import array from enum import Enum -from typing import Any, Dict, Optional, TypedDict -from typing import List -from dataclasses import field -from dataclasses import dataclass +from typing import Any, Dict, Optional, TypedDict, List +from dataclasses import dataclass, field from datetime import datetime - from .utils import try_get_from_dict - class GeocachingApiEnvironmentSettings(TypedDict): """Class to represent API environment settings""" - - api_scheme: str - api_host: str + api_scheme:str + api_host:str api_port: int - api_base_bath: str - + api_base_bath:str class GeocachingApiEnvironment(Enum): """Enum to represent API environment""" - - Staging = (1,) - Production = (2,) - + Staging = 1, + Production = 2, @dataclass class NearbyCachesSetting: location: GeocachingCoordinate radiusKm: float - class GeocachingSettings: """Class to hold the Geocaching Api settings""" - caches_codes: array(str) trackable_codes: array(str) environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__( - self, - environment: GeocachingApiEnvironment = GeocachingApiEnvironment.Production, - trackables: array(str) = [], - caches: array(str) = [], - nearby_caches_setting: NearbyCachesSetting = None, - ) -> None: + def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: array(str) = [], caches: array(str) = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.trackable_codes = trackables self.nearby_caches_setting = nearby_caches_setting @@ -61,11 +45,9 @@ def set_trackables(self, trackables: array(str)): def set_nearby_caches_setting(self, setting: NearbyCachesSetting): self.nearby_caches_setting = setting - @dataclass class GeocachingUser: """Class to hold the Geocaching user information""" - reference_code: Optional[str] = None username: Optional[str] = None find_count: Optional[int] = None @@ -77,30 +59,18 @@ class GeocachingUser: def update_from_dict(self, data: Dict[str, Any]) -> None: """Update user from the API result""" - self.reference_code = try_get_from_dict( - data, "referenceCode", self.reference_code - ) + self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.username = try_get_from_dict(data, "username", self.username) self.find_count = try_get_from_dict(data, "findCount", self.find_count) self.hide_count = try_get_from_dict(data, "hideCount", self.hide_count) - self.favorite_points = try_get_from_dict( - data, "favoritePoints", self.favorite_points - ) - self.souvenir_count = try_get_from_dict( - data, "souvenirCount", self.souvenir_count - ) - self.awarded_favorite_points = try_get_from_dict( - data, "awardedFavoritePoints", self.awarded_favorite_points - ) - self.membership_level_id = try_get_from_dict( - data, "membershipLevelId", self.membership_level_id - ) - + self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points) + self.souvenir_count = try_get_from_dict(data, "souvenirCount", self.souvenir_count) + self.awarded_favorite_points = try_get_from_dict(data, "awardedFavoritePoints", self.awarded_favorite_points) + self.membership_level_id = try_get_from_dict(data, "membershipLevelId", self.membership_level_id) @dataclass class GeocachingCoordinate: """Class to hold a Geocaching coordinate""" - latitude: Optional[str] = None longitude: Optional[str] = None @@ -109,11 +79,9 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: self.latitude = try_get_from_dict(data, "latitude", None) self.longitude = try_get_from_dict(data, "longitude", None) - @dataclass class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" - coordinates: GeocachingCoordinate = None logged_date: Optional[datetime] = None @@ -139,26 +107,21 @@ class GeocachingTrackableLog: logged_date: Optional[datetime] = None def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableLog: - self.reference_code = try_get_from_dict( - data, "referenceCode", self.reference_code - ) + self.reference_code = try_get_from_dict(data, 'referenceCode',self.reference_code) if self.owner is None: self.owner = GeocachingUser() - if "owner" in data: - self.owner.update_from_dict(data["owner"]) + if 'owner' in data: + self.owner.update_from_dict(data['owner']) else: self.owner = None - self.log_type = try_get_from_dict( - data["trackableLogType"], "name", self.log_type - ) - self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) - self.text = try_get_from_dict(data, "text", self.text) + self.log_type = try_get_from_dict(data['trackableLogType'], 'name',self.log_type) + self.logged_date = try_get_from_dict(data, 'loggedDate',self.logged_date) + self.text = try_get_from_dict(data, 'text',self.text) @dataclass class GeocachingTrackable: """Class to hold the Geocaching trackable information""" - reference_code: Optional[str] = None name: Optional[str] = None holder: GeocachingUser = None @@ -170,45 +133,31 @@ class GeocachingTrackable: latest_journey: Optional[GeocachingTrackableJourney] = None, trackable_journeys: Optional[List[GeocachingTrackableJourney]] = field(default_factory=list) - - is_missing: bool = (False,) - trackable_type: str = (None,) + is_missing: bool = False, + trackable_type: str = None, latest_log: GeocachingTrackableLog = None def update_from_dict(self, data: Dict[str, Any]) -> None: """Update trackable from the API""" - self.reference_code = try_get_from_dict( - data, "referenceCode", self.reference_code - ) + self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) if data["holder"] is not None: - if self.holder is None: + if self.holder is None : holder = GeocachingUser() holder.update_from_dict(data["holder"]) else: holder = None - self.tracking_number = try_get_from_dict( - data, "trackingNumber", self.tracking_number - ) - self.kilometers_traveled = try_get_from_dict( - data, "kilometersTraveled", self.kilometers_traveled, float - ) - self.miles_traveled = try_get_from_dict( - data, "milesTraveled", self.miles_traveled, float - ) - self.current_geocache_code = try_get_from_dict( - data, "currectGeocacheCode", self.current_geocache_code - ) - self.current_geocache_name = try_get_from_dict( - data, "currentGeocacheName", self.current_geocache_name - ) + self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) + self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) + self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled, float) + self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code) + self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name) self.is_missing = try_get_from_dict(data, "isMissing", self.is_missing) self.trackable_type = try_get_from_dict(data, "type", self.trackable_type) if "trackableLogs" in data and len(data["trackableLogs"]) > 0: self.latest_log = GeocachingTrackableLog(data=data["trackableLogs"][0]) - @dataclass class GeocachingCache: reference_code: Optional[str] = None @@ -217,9 +166,7 @@ class GeocachingCache: favoritePoints: Optional[int] = None def update_from_dict(self, data: Dict[str, Any]) -> None: - self.reference_code = try_get_from_dict( - data, "referenceCode", self.reference_code - ) + self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints) if "postedCoordinates" in data: @@ -227,10 +174,8 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: else: self.coordinates = None - class GeocachingStatus: """Class to hold all account status information""" - user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None nearby_caches: list[GeocachingCache] = None @@ -267,7 +212,7 @@ def update_trackables_from_dict(self, data: Any) -> None: if not reference_code in self.trackables.keys(): self.trackables[reference_code] = GeocachingTrackable() self.trackables[reference_code].update_from_dict(trackable) - + def update_nearby_caches_from_dict(self, data: Any) -> None: """Update nearby caches from the API result""" if not any(data): @@ -278,5 +223,6 @@ def update_nearby_caches_from_dict(self, data: Any) -> None: cache = GeocachingCache() cache.update_from_dict(cacheData) nearby_caches.append(cache) - + self.nearby_caches = nearby_caches + \ No newline at end of file From e98dd840afb3af0b20e88737251814e111ce447a Mon Sep 17 00:00:00 2001 From: marc7s Date: Sun, 24 Nov 2024 23:43:29 +0100 Subject: [PATCH 13/37] Fix errors and add missing cache data points --- geocachingapi/const.py | 10 ++++++++++ geocachingapi/geocachingapi.py | 18 ++++-------------- geocachingapi/models.py | 28 +++++++++++++++++----------- 3 files changed, 31 insertions(+), 25 deletions(-) diff --git a/geocachingapi/const.py b/geocachingapi/const.py index cf3f7e2..5fd6847 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -23,3 +23,13 @@ 2: "Charter", 3: "Premium" } + +CACHE_FIELDS_PARAMETER: str = ",".join([ + "referenceCode", + "name", + "postedCoordinates", + "favoritePoints", + #"findCount", # TODO: Include this another way, as this is not part of lite caches + "placedDate", + "location" + ]) \ No newline at end of file diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index bfb85d5..2a5e72f 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -12,7 +12,7 @@ from aiohttp import ClientResponse, ClientSession, ClientError from typing import Any, Awaitable, Callable, Dict, Optional -from .const import ENVIRONMENT_SETTINGS +from .const import ENVIRONMENT_SETTINGS, CACHE_FIELDS_PARAMETER from .exceptions import ( GeocachingApiConnectionError, GeocachingApiConnectionTimeoutError, @@ -145,7 +145,7 @@ async def update(self) -> GeocachingStatus: await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() - if self._settings.caches_codes is not None: + if len(self._settings.caches_codes) > 0: await self._get_cache_info() _LOGGER.info(f'Status updated.') @@ -154,13 +154,8 @@ async def update(self) -> GeocachingStatus: async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: - fields = ",".join([ - "name", - "postedCoordinates", - "favoritePoints" - ]) caches_parameters = ",".join(self._settings.caches_codes) - data = await self._request("GET",f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={fields}") + data = await self._request("GET", f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={CACHE_FIELDS_PARAMETER}") self._status.update_caches(data) _LOGGER.debug(f'Caches updated.') @@ -245,14 +240,9 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: return if data is None: - fields = ",".join([ - "referenceCode", - "name", - "postedCoordinates" - ]) coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location radiusKm: float = self._settings.nearby_caches_setting.radiusKm - URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusKm}km&fields={fields}&sort=distance+&lite=true" + URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusKm}km&fields={CACHE_FIELDS_PARAMETER}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually data = await self._request("GET", URL.replace("+", "%2B")) self._status.update_nearby_caches_from_dict(data) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index ce83979..c873584 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -1,5 +1,4 @@ from __future__ import annotations -from array import array from enum import Enum from typing import Any, Dict, Optional, TypedDict, List from dataclasses import dataclass, field @@ -25,22 +24,22 @@ class NearbyCachesSetting: class GeocachingSettings: """Class to hold the Geocaching Api settings""" - caches_codes: array(str) - trackable_codes: array(str) + caches_codes: list[str] + trackable_codes: list[str] environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: array(str) = [], caches: array(str) = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: list[str] = [], caches: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.trackable_codes = trackables self.nearby_caches_setting = nearby_caches_setting self.caches_codes = caches - def set_caches(self, caches: array(str)): - self.caches_codes = caches + def set_caches(self, cache_codes: list[str]): + self.caches_codes = cache_codes - def set_trackables(self, trackables: array(str)): - self.trackable_codes = trackables + def set_trackables(self, trackable_codes: list[str]): + self.trackable_codes = trackable_codes def set_nearby_caches_setting(self, setting: NearbyCachesSetting): self.nearby_caches_setting = setting @@ -164,11 +163,18 @@ class GeocachingCache: name: Optional[str] = None coordinates: GeocachingCoordinate = None favoritePoints: Optional[int] = None + findCount: Optional[int] = None + hiddenDate: Optional[datetime.date] = None + location: Optional[str] = None def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) - self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints) + self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints, int) + self.findCount = try_get_from_dict(data, "findCount", self.findCount, int) + self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, lambda d: datetime.date(datetime.fromisoformat(d))) + self.location = try_get_from_dict(data, "location", self.location) + if "postedCoordinates" in data: self.coordinates = GeocachingCoordinate(data=data["postedCoordinates"]) else: @@ -178,8 +184,8 @@ class GeocachingStatus: """Class to hold all account status information""" user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None - nearby_caches: list[GeocachingCache] = None - caches: GeocachingCache = None + nearby_caches: list[GeocachingCache] = [] + caches: list[GeocachingCache] = [] def __init__(self): """Initialize GeocachingStatus""" From aabbc7442cc3b1dc8655e4d46327f21c5ef2f715 Mon Sep 17 00:00:00 2001 From: marc7s Date: Sun, 24 Nov 2024 23:45:50 +0100 Subject: [PATCH 14/37] Rename caches -> tracked_caches --- geocachingapi/models.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index c873584..0b938f5 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -185,14 +185,14 @@ class GeocachingStatus: user: GeocachingUser = None trackables: Dict[str, GeocachingTrackable] = None nearby_caches: list[GeocachingCache] = [] - caches: list[GeocachingCache] = [] + tracked_caches: list[GeocachingCache] = [] def __init__(self): """Initialize GeocachingStatus""" self.user = GeocachingUser() self.trackables = {} self.nearby_caches = [] - self.caches = [] + self.tracked_caches = [] def update_user_from_dict(self, data: Dict[str, Any]) -> None: """Update user from the API result""" @@ -207,7 +207,7 @@ def update_caches(self, data: Any) -> None: cache = GeocachingCache() cache.update_from_dict(cacheData) caches.append(cache) - self.caches = caches + self.tracked_caches = caches def update_trackables_from_dict(self, data: Any) -> None: """Update trackables from the API result""" From c40377e6bb00c935f238eb08fed2d13b26008632 Mon Sep 17 00:00:00 2001 From: marc7s Date: Sun, 24 Nov 2024 23:55:40 +0100 Subject: [PATCH 15/37] Correctly parse location data for caches --- geocachingapi/models.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 0b938f5..c221f59 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -173,7 +173,13 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints, int) self.findCount = try_get_from_dict(data, "findCount", self.findCount, int) self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, lambda d: datetime.date(datetime.fromisoformat(d))) - self.location = try_get_from_dict(data, "location", self.location) + + # Parse the location + # Returns the location as "State, Country" if either could be parsed + location_obj: Dict[Any] = try_get_from_dict(data, "location", {}) + location_state: str = try_get_from_dict(location_obj, "state", "Unknown") + location_country: str = try_get_from_dict(location_obj, "country", "Unknown") + self.location = None if set([location_state, location_country]) == {"Unknown"} else f"{location_state}, {location_country}" if "postedCoordinates" in data: self.coordinates = GeocachingCoordinate(data=data["postedCoordinates"]) From 3176b7fffba7fb8fa13249ad9c684cc9a6272af9 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 25 Nov 2024 00:01:34 +0100 Subject: [PATCH 16/37] Rename caches_codes -> cache_codes --- geocachingapi/geocachingapi.py | 4 ++-- geocachingapi/models.py | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 2a5e72f..743601d 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -145,7 +145,7 @@ async def update(self) -> GeocachingStatus: await self._update_trackables() if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() - if len(self._settings.caches_codes) > 0: + if len(self._settings.cache_codes) > 0: await self._get_cache_info() _LOGGER.info(f'Status updated.') @@ -154,7 +154,7 @@ async def update(self) -> GeocachingStatus: async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: - caches_parameters = ",".join(self._settings.caches_codes) + caches_parameters = ",".join(self._settings.cache_codes) data = await self._request("GET", f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={CACHE_FIELDS_PARAMETER}") self._status.update_caches(data) _LOGGER.debug(f'Caches updated.') diff --git a/geocachingapi/models.py b/geocachingapi/models.py index c221f59..d38f660 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -24,7 +24,7 @@ class NearbyCachesSetting: class GeocachingSettings: """Class to hold the Geocaching Api settings""" - caches_codes: list[str] + cache_codes: list[str] trackable_codes: list[str] environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting @@ -33,10 +33,10 @@ def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironme """Initialize settings""" self.trackable_codes = trackables self.nearby_caches_setting = nearby_caches_setting - self.caches_codes = caches + self.cache_codes = caches def set_caches(self, cache_codes: list[str]): - self.caches_codes = cache_codes + self.cache_codes = cache_codes def set_trackables(self, trackable_codes: list[str]): self.trackable_codes = trackable_codes From 26be5b3a4474764ed52dce48a8cb3bf44f90f3b9 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 25 Nov 2024 16:25:52 +0100 Subject: [PATCH 17/37] Fix trackable parsing, add additonal fields to caches and trackables, some refactoring and cleanup --- geocachingapi/const.py | 1 + geocachingapi/geocachingapi.py | 22 +++++++------ geocachingapi/models.py | 56 +++++++++++++++++++--------------- 3 files changed, 46 insertions(+), 33 deletions(-) diff --git a/geocachingapi/const.py b/geocachingapi/const.py index 5fd6847..f9c1f90 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -27,6 +27,7 @@ CACHE_FIELDS_PARAMETER: str = ",".join([ "referenceCode", "name", + "owner", "postedCoordinates", "favoritePoints", #"findCount", # TODO: Include this another way, as this is not part of lite caches diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 743601d..51ed68b 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -139,8 +139,6 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: async def update(self) -> GeocachingStatus: await self._update_user() - if self._settings.trackable_codes is not None: - await self._update_trackable_journey() if len(self._settings.trackable_codes) > 0: await self._update_trackables() if self._settings.nearby_caches_setting is not None: @@ -194,16 +192,18 @@ async def _update_trackable_journey(self, data: Dict[str, Any] = None) -> None: trackable_parameters = ",".join(self._settings.trackable_codes) data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}") self._status.update_trackables_from_dict(data) + + # Update trackable journeys if len(self._status.trackables) > 0: for trackable in self._status.trackables.values(): - trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=1") + trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-") if trackable_journey_data: # Ensure data exists # Create a list of GeocachingTrackableJourney instances journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) for i, journey in enumerate(journeys): # Add each journey to the trackable's trackable_journeys list by index - trackable.trackable_journeys.append(journey) + trackable.journeys.append(journey) async def _update_trackables(self, data: Dict[str, Any] = None) -> None: assert self._status @@ -212,6 +212,8 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "referenceCode", "name", "holder", + "owner", + "releasedDate", "trackingNumber", "kilometersTraveled", "milesTraveled", @@ -223,13 +225,15 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: trackable_parameters = ",".join(self._settings.trackable_codes) data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}&expand=trackablelogs:1") self._status.update_trackables_from_dict(data) + + # Update trackable journeys if len(self._status.trackables) > 0: for trackable in self._status.trackables.values(): - latest_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-&take=1") - if len(latest_journey_data) == 1: - trackable.latest_journey = GeocachingTrackableJourney(data=latest_journey_data[0]) - else: - trackable.latest_journey = None + trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-") + if trackable_journey_data: + # Create a list of GeocachingTrackableJourney instances + journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) + trackable.journeys = journeys _LOGGER.debug(f'Trackables updated.') diff --git a/geocachingapi/models.py b/geocachingapi/models.py index d38f660..3991561 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -1,10 +1,22 @@ from __future__ import annotations from enum import Enum -from typing import Any, Dict, Optional, TypedDict, List +from typing import Any, Dict, Optional, TypedDict from dataclasses import dataclass, field from datetime import datetime from .utils import try_get_from_dict +DATETIME_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) + +def try_get_user_from_dict(data: Dict[str, Any], key: str, original_value: Any) -> GeocachingUser | None: + """Try to get user from dict, otherwise set default value""" + user_data = try_get_from_dict(data, key, None) + if user_data is None: + return original_value + + user = GeocachingUser() + user.update_from_dict(data[key]) + return user + class GeocachingApiEnvironmentSettings(TypedDict): """Class to represent API environment settings""" api_scheme:str @@ -93,8 +105,10 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) @classmethod - def from_list(cls, data_list: List[Dict[str, Any]]) -> List[GeocachingTrackableJourney]: + def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: """Creates a list of GeocachingTrackableJourney instances from an array of data""" + # TODO: Look into filtering this list for only journey-related logs + # Reference: https://api.groundspeak.com/documentation#trackable-log-types return [cls(data=data) for data in data_list] @dataclass @@ -106,16 +120,11 @@ class GeocachingTrackableLog: logged_date: Optional[datetime] = None def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableLog: - self.reference_code = try_get_from_dict(data, 'referenceCode',self.reference_code) - if self.owner is None: - self.owner = GeocachingUser() - if 'owner' in data: - self.owner.update_from_dict(data['owner']) - else: - self.owner = None - self.log_type = try_get_from_dict(data['trackableLogType'], 'name',self.log_type) - self.logged_date = try_get_from_dict(data, 'loggedDate',self.logged_date) - self.text = try_get_from_dict(data, 'text',self.text) + self.reference_code = try_get_from_dict(data, 'referenceCode', self.reference_code) + self.owner = try_get_user_from_dict(data, "owner", self.owner) + self.log_type = try_get_from_dict(data['trackableLogType'], 'name', self.log_type) + self.logged_date = try_get_from_dict(data, 'loggedDate', self.logged_date) + self.text = try_get_from_dict(data, 'text', self.text) @dataclass @@ -124,33 +133,30 @@ class GeocachingTrackable: reference_code: Optional[str] = None name: Optional[str] = None holder: GeocachingUser = None + owner: GeocachingUser = None + release_date: Optional[datetime.date] = None tracking_number: Optional[str] = None kilometers_traveled: Optional[float] = None miles_traveled: Optional[float] = None current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None - latest_journey: Optional[GeocachingTrackableJourney] = None, - trackable_journeys: Optional[List[GeocachingTrackableJourney]] = field(default_factory=list) + journeys: Optional[list[GeocachingTrackableJourney]] = field(default_factory=list) is_missing: bool = False, - trackable_type: str = None, + trackable_type: str = None latest_log: GeocachingTrackableLog = None def update_from_dict(self, data: Dict[str, Any]) -> None: """Update trackable from the API""" self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) - if data["holder"] is not None: - if self.holder is None : - holder = GeocachingUser() - holder.update_from_dict(data["holder"]) - else: - holder = None - + self.holder = try_get_user_from_dict(data, "holder", self.holder) + self.owner = try_get_user_from_dict(data, "owner", self.owner) + self.release_date = try_get_from_dict(data, "releasedDate", self.release_date, DATETIME_PARSER) self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled, float) - self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code) + self.current_geocache_code = try_get_from_dict(data, "currentGeocacheCode", self.current_geocache_code) self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name) self.is_missing = try_get_from_dict(data, "isMissing", self.is_missing) self.trackable_type = try_get_from_dict(data, "type", self.trackable_type) @@ -161,6 +167,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: class GeocachingCache: reference_code: Optional[str] = None name: Optional[str] = None + owner: GeocachingUser = None coordinates: GeocachingCoordinate = None favoritePoints: Optional[int] = None findCount: Optional[int] = None @@ -170,9 +177,10 @@ class GeocachingCache: def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) + self.owner = try_get_user_from_dict(data, "owner", self.owner) self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints, int) self.findCount = try_get_from_dict(data, "findCount", self.findCount, int) - self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, lambda d: datetime.date(datetime.fromisoformat(d))) + self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, DATETIME_PARSER) # Parse the location # Returns the location as "State, Country" if either could be parsed From 11f252c3759afe34b4575d850c883e93150cd8a9 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 25 Nov 2024 16:37:12 +0100 Subject: [PATCH 18/37] Formatting --- geocachingapi/models.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 3991561..b9e64b3 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -26,8 +26,8 @@ class GeocachingApiEnvironmentSettings(TypedDict): class GeocachingApiEnvironment(Enum): """Enum to represent API environment""" - Staging = 1, - Production = 2, + Staging = 1 + Production = 2 @dataclass class NearbyCachesSetting: @@ -120,11 +120,11 @@ class GeocachingTrackableLog: logged_date: Optional[datetime] = None def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableLog: - self.reference_code = try_get_from_dict(data, 'referenceCode', self.reference_code) + self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.owner = try_get_user_from_dict(data, "owner", self.owner) - self.log_type = try_get_from_dict(data['trackableLogType'], 'name', self.log_type) - self.logged_date = try_get_from_dict(data, 'loggedDate', self.logged_date) - self.text = try_get_from_dict(data, 'text', self.text) + self.log_type = try_get_from_dict(data["trackableLogType"], "name", self.log_type) + self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) + self.text = try_get_from_dict(data, "text", self.text) @dataclass From 71fd60f0a724067008f33746f3a1c109a75bd047 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 15:05:23 +0100 Subject: [PATCH 19/37] Always return original value if new value could not be parsed --- geocachingapi/utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/geocachingapi/utils.py b/geocachingapi/utils.py index a896666..8d816fb 100644 --- a/geocachingapi/utils.py +++ b/geocachingapi/utils.py @@ -4,7 +4,7 @@ def try_get_from_dict(data: Dict[str, Any], key: str, original_value: Any, conversion: Optional[Callable[[Any], Any]] = None) -> Any: """Try to get value from dict, otherwise set default value""" if not key in data: - return None + return original_value value = data[key] if value is None: From fdc0c4ab47c77371d97e866996d5edebd377f6bd Mon Sep 17 00:00:00 2001 From: Jakob Windt Date: Wed, 27 Nov 2024 16:04:35 +0100 Subject: [PATCH 20/37] Update NearbyCachesSetting --- geocachingapi/geocachingapi.py | 5 +++-- geocachingapi/models.py | 6 ++++++ 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 51ed68b..54d2216 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -245,8 +245,9 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: if data is None: coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location - radiusKm: float = self._settings.nearby_caches_setting.radiusKm - URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusKm}km&fields={CACHE_FIELDS_PARAMETER}&sort=distance+&lite=true" + radiusM: int = round(self._settings.nearby_caches_setting.radiusKm * 1000) + maxCount: int = self._settings.nearby_caches_setting.maxCount + URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually data = await self._request("GET", URL.replace("+", "%2B")) self._status.update_nearby_caches_from_dict(data) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index b9e64b3..59d3abb 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -33,6 +33,12 @@ class GeocachingApiEnvironment(Enum): class NearbyCachesSetting: location: GeocachingCoordinate radiusKm: float + maxCount: int + + def __init__(self, location: GeocachingCoordinate, radiusKm: float, maxCount: int) -> None: + self.location = location + self.radiusKm = radiusKm + self.maxCount = round(maxCount) class GeocachingSettings: """Class to hold the Geocaching Api settings""" From ee8343e01486ad44018544dcc56d2d4fe7b82288 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 16:32:20 +0100 Subject: [PATCH 21/37] Rename cache update function --- geocachingapi/geocachingapi.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 51ed68b..6f9b05b 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -144,16 +144,16 @@ async def update(self) -> GeocachingStatus: if self._settings.nearby_caches_setting is not None: await self._update_nearby_caches() if len(self._settings.cache_codes) > 0: - await self._get_cache_info() + await self._update_tracked_caches() _LOGGER.info(f'Status updated.') return self._status - async def _get_cache_info(self, data: Dict[str, Any] = None) -> None: + async def _update_tracked_caches(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: caches_parameters = ",".join(self._settings.cache_codes) - data = await self._request("GET", f"/geocaches?referenceCodes={caches_parameters}&lite=true&fields={CACHE_FIELDS_PARAMETER}") + data = await self._request("GET", f"/geocaches?referenceCodes={caches_parameters}&fields={CACHE_FIELDS_PARAMETER}&lite=true") self._status.update_caches(data) _LOGGER.debug(f'Caches updated.') From 50c7001c65f236f637a80e159f766d681e5b675b Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 16:34:03 +0100 Subject: [PATCH 22/37] Replace cache find count with found date and found switch --- geocachingapi/const.py | 2 +- geocachingapi/models.py | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/geocachingapi/const.py b/geocachingapi/const.py index f9c1f90..6a908f3 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -30,7 +30,7 @@ "owner", "postedCoordinates", "favoritePoints", - #"findCount", # TODO: Include this another way, as this is not part of lite caches + "userData", "placedDate", "location" ]) \ No newline at end of file diff --git a/geocachingapi/models.py b/geocachingapi/models.py index b9e64b3..e9239ac 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -170,17 +170,29 @@ class GeocachingCache: owner: GeocachingUser = None coordinates: GeocachingCoordinate = None favoritePoints: Optional[int] = None - findCount: Optional[int] = None hiddenDate: Optional[datetime.date] = None + foundDateTime: Optional[datetime] = None location: Optional[str] = None + @property + def found_by_user(self) -> bool: + return self.foundDateTime is not None + def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) self.owner = try_get_user_from_dict(data, "owner", self.owner) self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints, int) - self.findCount = try_get_from_dict(data, "findCount", self.findCount, int) self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, DATETIME_PARSER) + + # Parse the user data (information about this cache, specific to the user) + # The value is in data["userData"]["foundDate"], and is either None (not found) or a `datetime` object + if "userData" in data: + user_data_obj: Dict[Any] = try_get_from_dict(data, "userData", {}) + found_date_time: datetime | None = try_get_from_dict(user_data_obj, "foundDate", None, lambda d: None if d is None else datetime.fromisoformat(d)) + self.foundDateTime = found_date_time + else: + self.foundDateTime = None # Parse the location # Returns the location as "State, Country" if either could be parsed From 73c041f4289fc248e8749bfbaca0af132b404e87 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 17:00:34 +0100 Subject: [PATCH 23/37] Rework foundByUser to be nullable --- geocachingapi/models.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index e9239ac..133e7b8 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -172,12 +172,9 @@ class GeocachingCache: favoritePoints: Optional[int] = None hiddenDate: Optional[datetime.date] = None foundDateTime: Optional[datetime] = None + foundByUser: Optional[bool] = None location: Optional[str] = None - @property - def found_by_user(self) -> bool: - return self.foundDateTime is not None - def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) @@ -191,8 +188,10 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: user_data_obj: Dict[Any] = try_get_from_dict(data, "userData", {}) found_date_time: datetime | None = try_get_from_dict(user_data_obj, "foundDate", None, lambda d: None if d is None else datetime.fromisoformat(d)) self.foundDateTime = found_date_time + self.foundByUser = found_date_time is not None else: self.foundDateTime = None + self.foundByUser = None # Parse the location # Returns the location as "State, Country" if either could be parsed From ff85ef509d6c566dc324cb0e0c6ab55d77186c76 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 17:17:11 +0100 Subject: [PATCH 24/37] Fix nearby caches update condition and limit take parameter for API --- geocachingapi/geocachingapi.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 54d2216..25f76f0 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -141,7 +141,7 @@ async def update(self) -> GeocachingStatus: await self._update_user() if len(self._settings.trackable_codes) > 0: await self._update_trackables() - if self._settings.nearby_caches_setting is not None: + if self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.maxCount > 0: await self._update_nearby_caches() if len(self._settings.cache_codes) > 0: await self._get_cache_info() @@ -246,7 +246,7 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: if data is None: coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location radiusM: int = round(self._settings.nearby_caches_setting.radiusKm * 1000) - maxCount: int = self._settings.nearby_caches_setting.maxCount + maxCount: int = min(max(self._settings.nearby_caches_setting.maxCount, 0), 100) # Take range is 0-100 in API URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually data = await self._request("GET", URL.replace("+", "%2B")) From 412e898967bacb8296dcd4770f7e5f9e90dff0e1 Mon Sep 17 00:00:00 2001 From: marc7s Date: Wed, 27 Nov 2024 23:16:56 +0100 Subject: [PATCH 25/37] Remove unused function, align variables and functions to snake_case --- geocachingapi/geocachingapi.py | 49 ++++++----------------------- geocachingapi/models.py | 57 ++++++++++++++++++---------------- tests/test.py | 2 +- 3 files changed, 40 insertions(+), 68 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index bd671e1..90a6696 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -139,11 +139,11 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: async def update(self) -> GeocachingStatus: await self._update_user() - if len(self._settings.trackable_codes) > 0: + if len(self._settings.tracked_trackable_codes) > 0: await self._update_trackables() - if self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.maxCount > 0: + if self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.max_count > 0: await self._update_nearby_caches() - if len(self._settings.cache_codes) > 0: + if len(self._settings.tracked_cache_codes) > 0: await self._update_tracked_caches() _LOGGER.info(f'Status updated.') @@ -152,10 +152,10 @@ async def update(self) -> GeocachingStatus: async def _update_tracked_caches(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: - caches_parameters = ",".join(self._settings.cache_codes) - data = await self._request("GET", f"/geocaches?referenceCodes={caches_parameters}&fields={CACHE_FIELDS_PARAMETER}&lite=true") + cache_codes = ",".join(self._settings.tracked_cache_codes) + data = await self._request("GET", f"/geocaches?referenceCodes={cache_codes}&fields={CACHE_FIELDS_PARAMETER}&lite=true") self._status.update_caches(data) - _LOGGER.debug(f'Caches updated.') + _LOGGER.debug(f'Tracked caches updated.') async def _update_user(self, data: Dict[str, Any] = None) -> None: assert self._status @@ -173,37 +173,6 @@ async def _update_user(self, data: Dict[str, Any] = None) -> None: data = await self._request("GET", f"/users/me?fields={fields}") self._status.update_user_from_dict(data) _LOGGER.debug(f'User updated.') - - async def _update_trackable_journey(self, data: Dict[str, Any] = None) -> None: - assert self._status - if data is None: - fields = ",".join([ - "referenceCode", - "name", - "holder", - "trackingNumber", - "kilometersTraveled", - "milesTraveled", - "currentGeocacheCode", - "currentGeocacheName", - "isMissing", - "type" - ]) - trackable_parameters = ",".join(self._settings.trackable_codes) - data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}") - self._status.update_trackables_from_dict(data) - - # Update trackable journeys - if len(self._status.trackables) > 0: - for trackable in self._status.trackables.values(): - trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-") - if trackable_journey_data: # Ensure data exists - # Create a list of GeocachingTrackableJourney instances - journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) - - for i, journey in enumerate(journeys): - # Add each journey to the trackable's trackable_journeys list by index - trackable.journeys.append(journey) async def _update_trackables(self, data: Dict[str, Any] = None) -> None: assert self._status @@ -222,7 +191,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "isMissing", "type" ]) - trackable_parameters = ",".join(self._settings.trackable_codes) + trackable_parameters = ",".join(self._settings.tracked_trackable_codes) data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}&expand=trackablelogs:1") self._status.update_trackables_from_dict(data) @@ -245,8 +214,8 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: if data is None: coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location - radiusM: int = round(self._settings.nearby_caches_setting.radiusKm * 1000) - maxCount: int = min(max(self._settings.nearby_caches_setting.maxCount, 0), 100) # Take range is 0-100 in API + radiusM: int = round(self._settings.nearby_caches_setting.radius_km * 1000) + maxCount: int = min(max(self._settings.nearby_caches_setting.max_count, 0), 100) # Take range is 0-100 in API URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually data = await self._request("GET", URL.replace("+", "%2B")) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 80e5127..0eacd16 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -32,32 +32,32 @@ class GeocachingApiEnvironment(Enum): @dataclass class NearbyCachesSetting: location: GeocachingCoordinate - radiusKm: float - maxCount: int + radius_km: float + max_count: int def __init__(self, location: GeocachingCoordinate, radiusKm: float, maxCount: int) -> None: self.location = location - self.radiusKm = radiusKm - self.maxCount = round(maxCount) + self.radius_km = radiusKm + self.max_count = round(maxCount) class GeocachingSettings: """Class to hold the Geocaching Api settings""" - cache_codes: list[str] - trackable_codes: list[str] + tracked_cache_codes: list[str] + tracked_trackable_codes: list[str] environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables: list[str] = [], caches: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" - self.trackable_codes = trackables + self.tracked_trackable_codes = trackable_codes self.nearby_caches_setting = nearby_caches_setting - self.cache_codes = caches + self.tracked_cache_codes = cache_codes - def set_caches(self, cache_codes: list[str]): - self.cache_codes = cache_codes + def set_tracked_caches(self, cache_codes: list[str]): + self.tracked_cache_codes = cache_codes - def set_trackables(self, trackable_codes: list[str]): - self.trackable_codes = trackable_codes + def set_tracked_trackables(self, trackable_codes: list[str]): + self.tracked_trackable_codes = trackable_codes def set_nearby_caches_setting(self, setting: NearbyCachesSetting): self.nearby_caches_setting = setting @@ -166,6 +166,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.current_geocache_name = try_get_from_dict(data, "currentGeocacheName", self.current_geocache_name) self.is_missing = try_get_from_dict(data, "isMissing", self.is_missing) self.trackable_type = try_get_from_dict(data, "type", self.trackable_type) + if "trackableLogs" in data and len(data["trackableLogs"]) > 0: self.latest_log = GeocachingTrackableLog(data=data["trackableLogs"][0]) @@ -175,29 +176,29 @@ class GeocachingCache: name: Optional[str] = None owner: GeocachingUser = None coordinates: GeocachingCoordinate = None - favoritePoints: Optional[int] = None - hiddenDate: Optional[datetime.date] = None - foundDateTime: Optional[datetime] = None - foundByUser: Optional[bool] = None + favorite_points: Optional[int] = None + hidden_date: Optional[datetime.date] = None + found_date_time: Optional[datetime] = None + found_by_user: Optional[bool] = None location: Optional[str] = None def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) self.owner = try_get_user_from_dict(data, "owner", self.owner) - self.favoritePoints = try_get_from_dict(data, "favoritePoints", self.favoritePoints, int) - self.hiddenDate = try_get_from_dict(data, "placedDate", self.hiddenDate, DATETIME_PARSER) + self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points, int) + self.hidden_date = try_get_from_dict(data, "placedDate", self.hidden_date, DATETIME_PARSER) # Parse the user data (information about this cache, specific to the user) # The value is in data["userData"]["foundDate"], and is either None (not found) or a `datetime` object if "userData" in data: user_data_obj: Dict[Any] = try_get_from_dict(data, "userData", {}) found_date_time: datetime | None = try_get_from_dict(user_data_obj, "foundDate", None, lambda d: None if d is None else datetime.fromisoformat(d)) - self.foundDateTime = found_date_time - self.foundByUser = found_date_time is not None + self.found_date_time = found_date_time + self.found_by_user = found_date_time is not None else: - self.foundDateTime = None - self.foundByUser = None + self.found_date_time = None + self.found_by_user = None # Parse the location # Returns the location as "State, Country" if either could be parsed @@ -233,10 +234,11 @@ def update_caches(self, data: Any) -> None: """Update caches from the API result""" if not any(data): pass + caches: list[GeocachingCache] = [] - for cacheData in data: + for cache_data in data: cache = GeocachingCache() - cache.update_from_dict(cacheData) + cache.update_from_dict(cache_data) caches.append(cache) self.tracked_caches = caches @@ -244,6 +246,7 @@ def update_trackables_from_dict(self, data: Any) -> None: """Update trackables from the API result""" if not any(data): pass + for trackable in data: reference_code = trackable["referenceCode"] if not reference_code in self.trackables.keys(): @@ -256,9 +259,9 @@ def update_nearby_caches_from_dict(self, data: Any) -> None: pass nearby_caches: list[GeocachingCache] = [] - for cacheData in data: + for cache_data in data: cache = GeocachingCache() - cache.update_from_dict(cacheData) + cache.update_from_dict(cache_data) nearby_caches.append(cache) self.nearby_caches = nearby_caches diff --git a/tests/test.py b/tests/test.py index 20f930e..d3724a2 100644 --- a/tests/test.py +++ b/tests/test.py @@ -11,7 +11,7 @@ async def test(): """Function to test GeocachingAPI integration""" gc_settings = GeocachingSettings() api = GeocachingApi(token=TOKEN, environment=GeocachingApiEnvironment.Staging, settings=gc_settings) - gc_settings.set_trackables(['TB87DTF']) + gc_settings.set_tracked_trackables(['TB87DTF']) await api.update_settings(gc_settings) await _update(api) await api.close() From a4030e4fedacb6e653b086dbaf23138a2e60b322 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 28 Nov 2024 12:10:30 +0100 Subject: [PATCH 26/37] Improve trackable journey data and change API endpoint --- geocachingapi/geocachingapi.py | 22 +++++++++++++++++++++- geocachingapi/models.py | 23 ++++++++++++++--------- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 90a6696..f427571 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -198,12 +198,32 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: # Update trackable journeys if len(self._status.trackables) > 0: for trackable in self._status.trackables.values(): - trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/journeys?sort=loggedDate-") + fields = ",".join([ + "referenceCode", + "geocacheName", + "loggedDate", + "coordinates", + "url", + "owner" + ]) + max_log_count: int = 10 + + # Only fetch logs related to movement + # Reference: https://api.groundspeak.com/documentation#trackable-log-types + logTypes: list[int] = ",".join([ + "14", # Dropped Off + "15" # Transfer + ]) + trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/trackablelogs?fields={fields}&logTypes={logTypes}&take={max_log_count}") + if trackable_journey_data: # Create a list of GeocachingTrackableJourney instances journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) trackable.journeys = journeys + # Set the trackable coordinates to that of the latest log + trackable.coordinates = journeys[-1].coordinates + _LOGGER.debug(f'Trackables updated.') async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 0eacd16..ab54e81 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -5,7 +5,7 @@ from datetime import datetime from .utils import try_get_from_dict -DATETIME_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) +DATE_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) def try_get_user_from_dict(data: Dict[str, Any], key: str, original_value: Any) -> GeocachingUser | None: """Try to get user from dict, otherwise set default value""" @@ -100,22 +100,26 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" coordinates: GeocachingCoordinate = None - logged_date: Optional[datetime] = None + date: Optional[datetime] = None + user: GeocachingUser = None + cache_name: Optional[str] = None + url: Optional[str] = None def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: """Constructor for Geocaching trackable journey""" - if "coordinates" in data: + if "coordinates" in data and data["coordinates"] is not None: self.coordinates = GeocachingCoordinate(data=data["coordinates"]) else: self.coordinates = None - self.logged_date = try_get_from_dict(data, "loggedDate", self.logged_date) + self.date = try_get_from_dict(data, "loggedDate", self.date, DATE_PARSER) + self.user = try_get_user_from_dict(data, "owner", self.user) + self.cache_name = try_get_from_dict(data, "geocacheName", self.cache_name) + self.url = try_get_from_dict(data, "url", self.url) @classmethod def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: """Creates a list of GeocachingTrackableJourney instances from an array of data""" - # TODO: Look into filtering this list for only journey-related logs - # Reference: https://api.groundspeak.com/documentation#trackable-log-types - return [cls(data=data) for data in data_list] + return sorted([cls(data=data) for data in data_list], key=lambda j: j.date, reverse=False) @dataclass class GeocachingTrackableLog: @@ -147,6 +151,7 @@ class GeocachingTrackable: current_geocache_code: Optional[str] = None current_geocache_name: Optional[str] = None journeys: Optional[list[GeocachingTrackableJourney]] = field(default_factory=list) + coordinates: GeocachingCoordinate = None is_missing: bool = False, trackable_type: str = None @@ -158,7 +163,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.name = try_get_from_dict(data, "name", self.name) self.holder = try_get_user_from_dict(data, "holder", self.holder) self.owner = try_get_user_from_dict(data, "owner", self.owner) - self.release_date = try_get_from_dict(data, "releasedDate", self.release_date, DATETIME_PARSER) + self.release_date = try_get_from_dict(data, "releasedDate", self.release_date, DATE_PARSER) self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled, float) @@ -187,7 +192,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.name = try_get_from_dict(data, "name", self.name) self.owner = try_get_user_from_dict(data, "owner", self.owner) self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points, int) - self.hidden_date = try_get_from_dict(data, "placedDate", self.hidden_date, DATETIME_PARSER) + self.hidden_date = try_get_from_dict(data, "placedDate", self.hidden_date, DATE_PARSER) # Parse the user data (information about this cache, specific to the user) # The value is in data["userData"]["foundDate"], and is either None (not found) or a `datetime` object From 674f43fb0025914398ee847c97c54d4c2b34b776 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 28 Nov 2024 12:51:24 +0100 Subject: [PATCH 27/37] Reverse geocode trackable journey locations --- geocachingapi/models.py | 6 ++++++ setup.py | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index ab54e81..3e50537 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field from datetime import datetime from .utils import try_get_from_dict +import reverse_geocode DATE_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) @@ -100,6 +101,7 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" coordinates: GeocachingCoordinate = None + location_name: Optional[str] = None date: Optional[datetime] = None user: GeocachingUser = None cache_name: Optional[str] = None @@ -109,6 +111,10 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: """Constructor for Geocaching trackable journey""" if "coordinates" in data and data["coordinates"] is not None: self.coordinates = GeocachingCoordinate(data=data["coordinates"]) + location_info: dict[str, Any] = reverse_geocode.get((self.coordinates.latitude, self.coordinates.longitude)) + location_city: str = try_get_from_dict(location_info, "city", "Unknown") + location_country: str = try_get_from_dict(location_info, "country", "Unknown") + self.location_name = f"{location_city}, {location_country}" else: self.coordinates = None self.date = try_get_from_dict(data, "loggedDate", self.date, DATE_PARSER) diff --git a/setup.py b/setup.py index 4115567..7570ea3 100644 --- a/setup.py +++ b/setup.py @@ -23,7 +23,7 @@ def read(*parts): url="https://github.com/Sholofly/geocachingapi-python", packages=setuptools.find_packages(include=["geocachingapi"]), license="MIT license", - install_requires=["aiohttp>=3.7.4,<4", "backoff>=1.9.0", "yarl"], + install_requires=["aiohttp>=3.7.4,<4", "backoff>=1.9.0", "yarl", "reverse_geocode==1.6.5"], keywords=["geocaching", "api"], classifiers=[ "Development Status :: 3 - Alpha", From 45bf373f7b5e35ae1d02221c03ce99930a9f4504 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 28 Nov 2024 13:26:10 +0100 Subject: [PATCH 28/37] Add distance between journeys --- geocachingapi/geocachingapi.py | 17 +++++++++++++++++ geocachingapi/models.py | 11 +++++++++++ 2 files changed, 28 insertions(+) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index f427571..6230a02 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -219,6 +219,23 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: if trackable_journey_data: # Create a list of GeocachingTrackableJourney instances journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) + + # Calculate distances between journeys + # The journeys are sorted in order, so reverse it to iterate backwards + j_iter = iter(reversed(journeys)) + curr_journey: GeocachingTrackableJourney | None = next(j_iter) + prev_journey: GeocachingTrackableJourney | None = None + while True: + if curr_journey is None: + break + prev_journey = next(j_iter, None) + if prev_journey is None: + curr_journey.distance_km = 0 + break + + curr_journey.distance_km = GeocachingCoordinate.get_distance_km(prev_journey.coordinates, curr_journey.coordinates) + curr_journey = prev_journey + trackable.journeys = journeys # Set the trackable coordinates to that of the latest log diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 3e50537..6138b29 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -3,6 +3,7 @@ from typing import Any, Dict, Optional, TypedDict from dataclasses import dataclass, field from datetime import datetime +from math import radians, sin, cos, acos from .utils import try_get_from_dict import reverse_geocode @@ -97,11 +98,21 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: self.latitude = try_get_from_dict(data, "latitude", None) self.longitude = try_get_from_dict(data, "longitude", None) + def get_distance_km(coord1: GeocachingCoordinate, coord2: GeocachingCoordinate) -> float: + """Returns the distance in kilometers between two coordinates""" + mlat = radians(float(coord1.latitude)) + mlon = radians(float(coord1.longitude)) + plat = radians(float(coord2.latitude)) + plon = radians(float(coord2.longitude)) + earth_radius_km = 6371.01 + return earth_radius_km * acos(sin(mlat) * sin(plat) + cos(mlat) * cos(plat) * cos(mlon - plon)) + @dataclass class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" coordinates: GeocachingCoordinate = None location_name: Optional[str] = None + distance_km: Optional[float] = None date: Optional[datetime] = None user: GeocachingUser = None cache_name: Optional[str] = None From 7452276134a193d2c65f359aad3d7fcf2249ed99 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 28 Nov 2024 13:53:59 +0100 Subject: [PATCH 29/37] Handle blocking reverse geocoding outside of init function --- geocachingapi/geocachingapi.py | 2 +- geocachingapi/models.py | 23 ++++++++++++++++------- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 6230a02..be268fb 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -218,7 +218,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: if trackable_journey_data: # Create a list of GeocachingTrackableJourney instances - journeys = GeocachingTrackableJourney.from_list(trackable_journey_data) + journeys = await GeocachingTrackableJourney.from_list(trackable_journey_data) # Calculate distances between journeys # The journeys are sorted in order, so reverse it to iterate backwards diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 6138b29..4037d4c 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -6,6 +6,7 @@ from math import radians, sin, cos, acos from .utils import try_get_from_dict import reverse_geocode +import asyncio DATE_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) @@ -118,14 +119,11 @@ class GeocachingTrackableJourney: cache_name: Optional[str] = None url: Optional[str] = None - def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: + # Note: Reverse geocoding the journeys is not performed in the init function + def __init__(self, *, data: Dict[str, Any]) -> None: """Constructor for Geocaching trackable journey""" if "coordinates" in data and data["coordinates"] is not None: self.coordinates = GeocachingCoordinate(data=data["coordinates"]) - location_info: dict[str, Any] = reverse_geocode.get((self.coordinates.latitude, self.coordinates.longitude)) - location_city: str = try_get_from_dict(location_info, "city", "Unknown") - location_country: str = try_get_from_dict(location_info, "country", "Unknown") - self.location_name = f"{location_city}, {location_country}" else: self.coordinates = None self.date = try_get_from_dict(data, "loggedDate", self.date, DATE_PARSER) @@ -134,9 +132,20 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: self.url = try_get_from_dict(data, "url", self.url) @classmethod - def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: + async def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: """Creates a list of GeocachingTrackableJourney instances from an array of data""" - return sorted([cls(data=data) for data in data_list], key=lambda j: j.date, reverse=False) + journeys: list[GeocachingTrackableJourney] = sorted([cls(data=data) for data in data_list], key=lambda j: j.date, reverse=False) + + # Reverse geocoding the journey locations reads from a file and is therefore a blocking call + # Therefore, we go over all journeys and perform the reverse geocoding pass after they have been initialized + loop = asyncio.get_running_loop() + for journey in journeys: + location_info: dict[str, Any] = await loop.run_in_executor(None, reverse_geocode.get, (journey.coordinates.latitude, journey.coordinates.longitude)) + location_city: str = try_get_from_dict(location_info, "city", "Unknown") + location_country: str = try_get_from_dict(location_info, "country", "Unknown") + journey.location_name = f"{location_city}, {location_country}" + + return journeys @dataclass class GeocachingTrackableLog: From 9a457b7dfbd95579da1aef5fc292eacf779c2afc Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 5 Dec 2024 15:20:57 +0100 Subject: [PATCH 30/37] Separate nearby caches logic to allow directed usage --- geocachingapi/geocachingapi.py | 25 ++++++++++++++++++------- geocachingapi/models.py | 24 +++++++++++++----------- 2 files changed, 31 insertions(+), 18 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index be268fb..5e86a2b 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -21,6 +21,7 @@ ) from .models import ( + GeocachingCache, GeocachingCoordinate, GeocachingStatus, GeocachingSettings, @@ -250,15 +251,25 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: return if data is None: - coordinates: GeocachingCoordinate = self._settings.nearby_caches_setting.location - radiusM: int = round(self._settings.nearby_caches_setting.radius_km * 1000) - maxCount: int = min(max(self._settings.nearby_caches_setting.max_count, 0), 100) # Take range is 0-100 in API - URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" - # The + sign is not encoded correctly, so we encode it manually - data = await self._request("GET", URL.replace("+", "%2B")) - self._status.update_nearby_caches_from_dict(data) + self._status.nearby_caches = await self.get_nearby_caches( + self._settings.nearby_caches_setting.location, + self._settings.nearby_caches_setting.radius_km, + self._settings.nearby_caches_setting.max_count + ) + else: + self._status.update_nearby_caches_from_dict(data) _LOGGER.debug(f'Nearby caches updated.') + + async def get_nearby_caches(self, coordinates: GeocachingCoordinate, radius_km: float, max_count: int = 10) -> list[GeocachingCache]: + radiusM: int = round(radius_km * 1000) + maxCount: int = min(max(max_count, 0), 100) # Take range is 0-100 in API + + URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" + # The + sign is not encoded correctly, so we encode it manually + data = await self._request("GET", URL.replace("+", "%2B")) + + return GeocachingStatus.parse_caches(data) async def update_settings(self, settings: GeocachingSettings): """Update the Geocaching settings""" diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 4037d4c..37672d1 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -265,13 +265,8 @@ def update_caches(self, data: Any) -> None: """Update caches from the API result""" if not any(data): pass - - caches: list[GeocachingCache] = [] - for cache_data in data: - cache = GeocachingCache() - cache.update_from_dict(cache_data) - caches.append(cache) - self.tracked_caches = caches + + self.tracked_caches = GeocachingStatus.parse_caches(data) def update_trackables_from_dict(self, data: Any) -> None: """Update trackables from the API result""" @@ -289,11 +284,18 @@ def update_nearby_caches_from_dict(self, data: Any) -> None: if not any(data): pass - nearby_caches: list[GeocachingCache] = [] + self.nearby_caches = GeocachingStatus.parse_caches(data) + + @staticmethod + def parse_caches(data: Any) -> list[GeocachingCache]: + """Parse caches from the API result""" + if data is None: + return [] + + caches: list[GeocachingCache] = [] for cache_data in data: cache = GeocachingCache() cache.update_from_dict(cache_data) - nearby_caches.append(cache) + caches.append(cache) - self.nearby_caches = nearby_caches - \ No newline at end of file + return caches \ No newline at end of file From 75285e83f60aec43642eaa2b4f1452b891239a31 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 9 Dec 2024 15:02:03 +0100 Subject: [PATCH 31/37] Add comments and improve documentation --- geocachingapi/const.py | 1 + geocachingapi/geocachingapi.py | 18 +++++++++++- geocachingapi/models.py | 54 +++++++++++++++++++--------------- geocachingapi/utils.py | 10 +++---- 4 files changed, 54 insertions(+), 29 deletions(-) diff --git a/geocachingapi/const.py b/geocachingapi/const.py index 6a908f3..0e0c3b4 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -24,6 +24,7 @@ 3: "Premium" } +# Required parameters for fetching caches in order to generate complete GeocachingCache objects CACHE_FIELDS_PARAMETER: str = ",".join([ "referenceCode", "name", diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 5e86a2b..ff79a7e 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -139,11 +139,18 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: return result async def update(self) -> GeocachingStatus: + # First, update the user await self._update_user() + + # If we are tracking trackables, update them if len(self._settings.tracked_trackable_codes) > 0: await self._update_trackables() + + # If the nearby caches setting is enabled, update them if self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.max_count > 0: await self._update_nearby_caches() + + # If we are tracking caches, update them if len(self._settings.tracked_cache_codes) > 0: await self._update_tracked_caches() @@ -217,6 +224,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: ]) trackable_journey_data = await self._request("GET",f"/trackables/{trackable.reference_code}/trackablelogs?fields={fields}&logTypes={logTypes}&take={max_log_count}") + # Note that if we are not fetching all journeys, the distance for the first journey in our data will be incorrect, since it does not know there was a previous journey if trackable_journey_data: # Create a list of GeocachingTrackableJourney instances journeys = await GeocachingTrackableJourney.from_list(trackable_journey_data) @@ -224,16 +232,22 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: # Calculate distances between journeys # The journeys are sorted in order, so reverse it to iterate backwards j_iter = iter(reversed(journeys)) + + # Since we are iterating backwards, next is actually the previous journey. + # However, the previous journey is set in the loop, so we assume it is missing for now curr_journey: GeocachingTrackableJourney | None = next(j_iter) prev_journey: GeocachingTrackableJourney | None = None while True: + # Ensure that the current journey is valid if curr_journey is None: break prev_journey = next(j_iter, None) + # If we have reached the first journey, its distance should be 0 (it did not travel from anywhere) if prev_journey is None: curr_journey.distance_km = 0 break + # Calculate the distance from the previous to the current location, as that is the distance the current journey travelled curr_journey.distance_km = GeocachingCoordinate.get_distance_km(prev_journey.coordinates, curr_journey.coordinates) curr_journey = prev_journey @@ -245,6 +259,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: _LOGGER.debug(f'Trackables updated.') async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: + """Update the nearby caches""" assert self._status if self._settings.nearby_caches_setting is None: _LOGGER.warning("Cannot update nearby caches, setting has not been configured.") @@ -262,7 +277,8 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: _LOGGER.debug(f'Nearby caches updated.') async def get_nearby_caches(self, coordinates: GeocachingCoordinate, radius_km: float, max_count: int = 10) -> list[GeocachingCache]: - radiusM: int = round(radius_km * 1000) + """Get caches nearby the provided coordinates, within the provided radius""" + radiusM: int = round(radius_km * 1000) # Convert the radius from km to m maxCount: int = min(max(max_count, 0), 100) # Take range is 0-100 in API URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 37672d1..51b4194 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -8,6 +8,7 @@ import reverse_geocode import asyncio +# Parser that parses an ISO date string to a date (not datetime) DATE_PARSER = lambda d: datetime.date(datetime.fromisoformat(d)) def try_get_user_from_dict(data: Dict[str, Any], key: str, original_value: Any) -> GeocachingUser | None: @@ -34,27 +35,28 @@ class GeocachingApiEnvironment(Enum): @dataclass class NearbyCachesSetting: - location: GeocachingCoordinate - radius_km: float - max_count: int + """Class to hold the nearby caches settings, as part of the API settings""" + location: GeocachingCoordinate # The position from which to search for nearby caches + radius_km: float # The radius around the position to search + max_count: int # The max number of nearby caches to return def __init__(self, location: GeocachingCoordinate, radiusKm: float, maxCount: int) -> None: self.location = location self.radius_km = radiusKm - self.max_count = round(maxCount) + self.max_count = max(0, round(maxCount)) class GeocachingSettings: - """Class to hold the Geocaching Api settings""" + """Class to hold the Geocaching API settings""" tracked_cache_codes: list[str] tracked_trackable_codes: list[str] environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__(self, environment: GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.tracked_trackable_codes = trackable_codes - self.nearby_caches_setting = nearby_caches_setting self.tracked_cache_codes = cache_codes + self.nearby_caches_setting = nearby_caches_setting def set_tracked_caches(self, cache_codes: list[str]): self.tracked_cache_codes = cache_codes @@ -100,26 +102,26 @@ def __init__(self, *, data: Dict[str, Any]) -> GeocachingCoordinate: self.longitude = try_get_from_dict(data, "longitude", None) def get_distance_km(coord1: GeocachingCoordinate, coord2: GeocachingCoordinate) -> float: - """Returns the distance in kilometers between two coordinates""" - mlat = radians(float(coord1.latitude)) - mlon = radians(float(coord1.longitude)) - plat = radians(float(coord2.latitude)) - plon = radians(float(coord2.longitude)) - earth_radius_km = 6371.01 + """Returns the distance in kilometers between two coordinates. Returns the great-circle distance between the coordinates""" + mlat: float = radians(float(coord1.latitude)) + mlon: float = radians(float(coord1.longitude)) + plat: float = radians(float(coord2.latitude)) + plon: float = radians(float(coord2.longitude)) + earth_radius_km: float = 6371.01 return earth_radius_km * acos(sin(mlat) * sin(plat) + cos(mlat) * cos(plat) * cos(mlon - plon)) @dataclass class GeocachingTrackableJourney: """Class to hold Geocaching trackable journey information""" - coordinates: GeocachingCoordinate = None - location_name: Optional[str] = None - distance_km: Optional[float] = None - date: Optional[datetime] = None - user: GeocachingUser = None - cache_name: Optional[str] = None - url: Optional[str] = None - - # Note: Reverse geocoding the journeys is not performed in the init function + coordinates: GeocachingCoordinate = None # The location at the end of this journey + location_name: Optional[str] = None # A reverse geocoded name of the location at the end of this journey + distance_km: Optional[float] = None # The distance the trackable travelled in this journey + date: Optional[datetime] = None # The date when this journey was completed + user: GeocachingUser = None # The Geocaching user who moved the trackable during this journey + cache_name: Optional[str] = None # The name of the cache the trackable resided in at the end of this journey + url: Optional[str] = None # A link to this journey + + # Note: Reverse geocoding the journeys is not performed in the init function as it is an asynchronous operation def __init__(self, *, data: Dict[str, Any]) -> None: """Constructor for Geocaching trackable journey""" if "coordinates" in data and data["coordinates"] is not None: @@ -133,16 +135,21 @@ def __init__(self, *, data: Dict[str, Any]) -> None: @classmethod async def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: - """Creates a list of GeocachingTrackableJourney instances from an array of data""" + """Creates a list of GeocachingTrackableJourney instances from an array of data, in order from oldest to newest""" journeys: list[GeocachingTrackableJourney] = sorted([cls(data=data) for data in data_list], key=lambda j: j.date, reverse=False) # Reverse geocoding the journey locations reads from a file and is therefore a blocking call # Therefore, we go over all journeys and perform the reverse geocoding pass after they have been initialized loop = asyncio.get_running_loop() for journey in journeys: + # Get the location information from the `reverse_geocode` package location_info: dict[str, Any] = await loop.run_in_executor(None, reverse_geocode.get, (journey.coordinates.latitude, journey.coordinates.longitude)) + + # Parse the response to extract the relevant data location_city: str = try_get_from_dict(location_info, "city", "Unknown") location_country: str = try_get_from_dict(location_info, "country", "Unknown") + + # Set the location name to a formatted string journey.location_name = f"{location_city}, {location_country}" return journeys @@ -236,6 +243,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: location_obj: Dict[Any] = try_get_from_dict(data, "location", {}) location_state: str = try_get_from_dict(location_obj, "state", "Unknown") location_country: str = try_get_from_dict(location_obj, "country", "Unknown") + # Set the location to `None` if both state and country are unknown, otherwise set it to the known data self.location = None if set([location_state, location_country]) == {"Unknown"} else f"{location_state}, {location_country}" if "postedCoordinates" in data: diff --git a/geocachingapi/utils.py b/geocachingapi/utils.py index 8d816fb..35b3e39 100644 --- a/geocachingapi/utils.py +++ b/geocachingapi/utils.py @@ -1,14 +1,14 @@ """Utils for consuming the API""" from typing import Dict, Any, Callable, Optional -def try_get_from_dict(data: Dict[str, Any], key: str, original_value: Any, conversion: Optional[Callable[[Any], Any]] = None) -> Any: - """Try to get value from dict, otherwise set default value""" +def try_get_from_dict(data: Dict[str, Any], key: str, fallback: Any, conversion: Optional[Callable[[Any], Any]] = None) -> Any: + """Try to get value from dict, otherwise return fallback value""" if not key in data: - return original_value + return fallback value = data[key] if value is None: - return original_value + return fallback if conversion is None: - return value + return value return conversion(value) From 411aa0a1e94acee1dfdb8dc663c38fd668e27fe0 Mon Sep 17 00:00:00 2001 From: marc7s Date: Mon, 9 Dec 2024 21:09:20 +0100 Subject: [PATCH 32/37] Add settings validation method --- geocachingapi/exceptions.py | 7 +++++- geocachingapi/geocachingapi.py | 44 +++++++++++++++++++++++++++++----- 2 files changed, 44 insertions(+), 7 deletions(-) diff --git a/geocachingapi/exceptions.py b/geocachingapi/exceptions.py index 79d65a2..7aa6eda 100644 --- a/geocachingapi/exceptions.py +++ b/geocachingapi/exceptions.py @@ -1,8 +1,13 @@ -"""Exceptions for the Gecaching API.""" +"""Exceptions for the Geocaching API.""" class GeocachingApiError(Exception): """Generic GeocachingApi exception.""" +class GeocachingInvalidSettingsError(Exception): + """GeocachingApi invalid settings exception.""" + def __init__(self, code_type: str, invalid_codes: set[str]): + super().__init__(f"Invalid {code_type} codes: {', '.join(invalid_codes)}") + class GeocachingApiConnectionError(GeocachingApiError): """GeocachingApi connection exception.""" diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index ff79a7e..8f526c1 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -18,6 +18,7 @@ GeocachingApiConnectionTimeoutError, GeocachingApiError, GeocachingApiRateLimitError, + GeocachingInvalidSettingsError, ) from .models import ( @@ -137,22 +138,31 @@ async def _request(self, method, uri, **kwargs) -> ClientResponse: _LOGGER.debug(f'Response:') _LOGGER.debug(f'{str(result)}') return result + + def _tracked_trackables_enabled(self) -> bool: + return len(self._settings.tracked_trackable_codes) > 0 + + def _tracked_caches_enabled(self) -> bool: + return len(self._settings.tracked_cache_codes) > 0 + + def _nearby_caches_enabled(self) -> bool: + return self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.max_count > 0 async def update(self) -> GeocachingStatus: # First, update the user await self._update_user() # If we are tracking trackables, update them - if len(self._settings.tracked_trackable_codes) > 0: + if self._tracked_trackables_enabled(): await self._update_trackables() - # If the nearby caches setting is enabled, update them - if self._settings.nearby_caches_setting is not None and self._settings.nearby_caches_setting.max_count > 0: - await self._update_nearby_caches() - # If we are tracking caches, update them - if len(self._settings.tracked_cache_codes) > 0: + if self._tracked_caches_enabled(): await self._update_tracked_caches() + + # If the nearby caches setting is enabled, update them + if self._nearby_caches_enabled(): + await self._update_nearby_caches() _LOGGER.info(f'Status updated.') return self._status @@ -286,6 +296,28 @@ async def get_nearby_caches(self, coordinates: GeocachingCoordinate, radius_km: data = await self._request("GET", URL.replace("+", "%2B")) return GeocachingStatus.parse_caches(data) + + async def _verify_codes(self, endpoint: str, code_type: str, reference_codes: list[str], extra_params: dict[str, str] = {}) -> None: + """Verifies a set of reference codes to ensure they are valid, and returns a set of all invalid codes""" + ref_codes_param: str = ",".join(reference_codes) + additional_params: str = "&".join([f'{name}={val}' for name, val in extra_params.items()]) + additional_params = "&" + additional_params if len(additional_params) > 0 else "" + + data = await self._request("GET", f"/{endpoint}?referenceCodes={ref_codes_param}&fields=referenceCode{additional_params}") + invalid_codes: set[str] = set(reference_codes).difference([d["referenceCode"] for d in data]) + + if len(invalid_codes) > 0: + raise GeocachingInvalidSettingsError(code_type, invalid_codes) + + async def verify_settings(self) -> None: + """Verifies the settings, checking for invalid reference codes""" + # Verify the tracked trackable reference codes + if self._tracked_trackables_enabled(): + await self._verify_codes("trackables", "trackable", self._settings.tracked_trackable_codes) + + # Verify the tracked cache reference codes + if self._tracked_caches_enabled(): + await self._verify_codes("geocaches", "geocache", self._settings.tracked_cache_codes, {"lite": "true"}) async def update_settings(self, settings: GeocachingSettings): """Update the Geocaching settings""" From 7586c3d8431320855bea0a5cf37f0caec6e94723 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 12 Dec 2024 15:38:16 +0100 Subject: [PATCH 33/37] Add cache and trackable URLs --- geocachingapi/const.py | 1 + geocachingapi/geocachingapi.py | 1 + geocachingapi/models.py | 4 ++++ 3 files changed, 6 insertions(+) diff --git a/geocachingapi/const.py b/geocachingapi/const.py index 0e0c3b4..18334a2 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -30,6 +30,7 @@ "name", "owner", "postedCoordinates", + "url", "favoritePoints", "userData", "placedDate", diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 8f526c1..0e804cf 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -200,6 +200,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "name", "holder", "owner", + "url", "releasedDate", "trackingNumber", "kilometersTraveled", diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 51b4194..1002163 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -177,6 +177,7 @@ class GeocachingTrackable: name: Optional[str] = None holder: GeocachingUser = None owner: GeocachingUser = None + url: Optional[str] = None release_date: Optional[datetime.date] = None tracking_number: Optional[str] = None kilometers_traveled: Optional[float] = None @@ -196,6 +197,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.name = try_get_from_dict(data, "name", self.name) self.holder = try_get_user_from_dict(data, "holder", self.holder) self.owner = try_get_user_from_dict(data, "owner", self.owner) + self.url = try_get_from_dict(data, "url", self.url) self.release_date = try_get_from_dict(data, "releasedDate", self.release_date, DATE_PARSER) self.tracking_number = try_get_from_dict(data, "trackingNumber", self.tracking_number) self.kilometers_traveled = try_get_from_dict(data, "kilometersTraveled", self.kilometers_traveled, float) @@ -214,6 +216,7 @@ class GeocachingCache: name: Optional[str] = None owner: GeocachingUser = None coordinates: GeocachingCoordinate = None + url: Optional[str] = None favorite_points: Optional[int] = None hidden_date: Optional[datetime.date] = None found_date_time: Optional[datetime] = None @@ -224,6 +227,7 @@ def update_from_dict(self, data: Dict[str, Any]) -> None: self.reference_code = try_get_from_dict(data, "referenceCode", self.reference_code) self.name = try_get_from_dict(data, "name", self.name) self.owner = try_get_user_from_dict(data, "owner", self.owner) + self.url = try_get_from_dict(data, "url", self.url) self.favorite_points = try_get_from_dict(data, "favoritePoints", self.favorite_points, int) self.hidden_date = try_get_from_dict(data, "placedDate", self.hidden_date, DATE_PARSER) From 2524b003caa9135e724e4b0ca2c7b99b945b899e Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 12 Dec 2024 16:55:27 +0100 Subject: [PATCH 34/37] Formatting and cleanup --- geocachingapi/geocachingapi.py | 3 ++- geocachingapi/models.py | 10 +++++----- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 0e804cf..91d5800 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -53,7 +53,7 @@ def __init__( """Initialize connection with the Geocaching API.""" self._environment_settings = ENVIRONMENT_SETTINGS[environment] self._status = GeocachingStatus() - self._settings = settings or GeocachingSettings(False) + self._settings = settings or GeocachingSettings() self._session = session self.request_timeout = request_timeout self.token = token @@ -172,6 +172,7 @@ async def _update_tracked_caches(self, data: Dict[str, Any] = None) -> None: if data is None: cache_codes = ",".join(self._settings.tracked_cache_codes) data = await self._request("GET", f"/geocaches?referenceCodes={cache_codes}&fields={CACHE_FIELDS_PARAMETER}&lite=true") + self._status.update_caches(data) _LOGGER.debug(f'Tracked caches updated.') diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 1002163..d42a2ca 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -23,10 +23,10 @@ def try_get_user_from_dict(data: Dict[str, Any], key: str, original_value: Any) class GeocachingApiEnvironmentSettings(TypedDict): """Class to represent API environment settings""" - api_scheme:str - api_host:str + api_scheme: str + api_host: str api_port: int - api_base_bath:str + api_base_bath: str class GeocachingApiEnvironment(Enum): """Enum to represent API environment""" @@ -52,7 +52,7 @@ class GeocachingSettings: environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment: GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__(self, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.tracked_trackable_codes = trackable_codes self.tracked_cache_codes = cache_codes @@ -187,7 +187,7 @@ class GeocachingTrackable: journeys: Optional[list[GeocachingTrackableJourney]] = field(default_factory=list) coordinates: GeocachingCoordinate = None - is_missing: bool = False, + is_missing: bool = False trackable_type: str = None latest_log: GeocachingTrackableLog = None From 074fad6311cf411dee25982492877c923d5ca3d4 Mon Sep 17 00:00:00 2001 From: marc7s Date: Thu, 12 Dec 2024 18:03:23 +0100 Subject: [PATCH 35/37] Update test --- tests/test.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/test.py b/tests/test.py index d3724a2..dee7225 100644 --- a/tests/test.py +++ b/tests/test.py @@ -29,10 +29,11 @@ async def _update(api:GeocachingApi): print(f'Kilometers traveled: {trackable.kilometers_traveled}km') print(f'Miles traveled: {trackable.miles_traveled}mi') print(f'Missing?: {trackable.is_missing}') - if trackable.latest_journey: - print(f'last journey: {trackable.latest_journey.logged_date}') - print(f'latitude: {trackable.latest_journey.coordinates.latitude}') - print(f'longitude: {trackable.latest_journey.coordinates.longitude}') + if trackable.journeys and len(trackable.journeys) > 0: + latest_journey = trackable.journeys[-1] + print(f'last journey: {latest_journey.date}') + print(f'latitude: {latest_journey.coordinates.latitude}') + print(f'longitude: {latest_journey.coordinates.longitude}') if trackable.latest_log: print(f'last log date: {trackable.latest_log.logged_date}') print(f'last log type: {trackable.latest_log.log_type}') From 08b5a4f870bce555ac103ce8542a42a96ddaa514 Mon Sep 17 00:00:00 2001 From: marc7s Date: Sat, 14 Dec 2024 01:02:07 +0100 Subject: [PATCH 36/37] Limits: cache and trackable limits in settings and in API call. Error handling: Raise error if too many codes were configured in settings. Automatically remove duplicate codes --- geocachingapi/exceptions.py | 4 ++++ geocachingapi/geocachingapi.py | 32 +++++++++++++++++--------------- geocachingapi/models.py | 18 +++++++++++++----- geocachingapi/utils.py | 4 ++++ 4 files changed, 38 insertions(+), 20 deletions(-) diff --git a/geocachingapi/exceptions.py b/geocachingapi/exceptions.py index 7aa6eda..104c787 100644 --- a/geocachingapi/exceptions.py +++ b/geocachingapi/exceptions.py @@ -8,6 +8,10 @@ class GeocachingInvalidSettingsError(Exception): def __init__(self, code_type: str, invalid_codes: set[str]): super().__init__(f"Invalid {code_type} codes: {', '.join(invalid_codes)}") +class GeocachingTooManyCodesError(GeocachingApiError): + """GeocachingApi settings exception: too many codes.""" + def __init__(self, message: str): + super().__init__(message) class GeocachingApiConnectionError(GeocachingApiError): """GeocachingApi connection exception.""" diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index 91d5800..ec8d45d 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -20,6 +20,7 @@ GeocachingApiRateLimitError, GeocachingInvalidSettingsError, ) +from .utils import clamp from .models import ( GeocachingCache, @@ -167,15 +168,6 @@ async def update(self) -> GeocachingStatus: _LOGGER.info(f'Status updated.') return self._status - async def _update_tracked_caches(self, data: Dict[str, Any] = None) -> None: - assert self._status - if data is None: - cache_codes = ",".join(self._settings.tracked_cache_codes) - data = await self._request("GET", f"/geocaches?referenceCodes={cache_codes}&fields={CACHE_FIELDS_PARAMETER}&lite=true") - - self._status.update_caches(data) - _LOGGER.debug(f'Tracked caches updated.') - async def _update_user(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: @@ -192,6 +184,15 @@ async def _update_user(self, data: Dict[str, Any] = None) -> None: data = await self._request("GET", f"/users/me?fields={fields}") self._status.update_user_from_dict(data) _LOGGER.debug(f'User updated.') + + async def _update_tracked_caches(self, data: Dict[str, Any] = None) -> None: + assert self._status + if data is None: + cache_codes = ",".join(self._settings.tracked_cache_codes) + data = await self._request("GET", f"/geocaches?referenceCodes={cache_codes}&fields={CACHE_FIELDS_PARAMETER}&lite=true") + + self._status.update_caches(data) + _LOGGER.debug(f'Tracked caches updated.') async def _update_trackables(self, data: Dict[str, Any] = None) -> None: assert self._status @@ -212,7 +213,8 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "type" ]) trackable_parameters = ",".join(self._settings.tracked_trackable_codes) - data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}&expand=trackablelogs:1") + max_count_param: int = clamp(len(self._settings.tracked_trackable_codes), 0, 50) # Take range is 0-50 in API + data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}&take={max_count_param}&expand=trackablelogs:1") self._status.update_trackables_from_dict(data) # Update trackable journeys @@ -226,7 +228,7 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "url", "owner" ]) - max_log_count: int = 10 + max_log_count: int = clamp(10, 0, 50) # Take range is 0-50 in API # Only fetch logs related to movement # Reference: https://api.groundspeak.com/documentation#trackable-log-types @@ -291,22 +293,22 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: async def get_nearby_caches(self, coordinates: GeocachingCoordinate, radius_km: float, max_count: int = 10) -> list[GeocachingCache]: """Get caches nearby the provided coordinates, within the provided radius""" radiusM: int = round(radius_km * 1000) # Convert the radius from km to m - maxCount: int = min(max(max_count, 0), 100) # Take range is 0-100 in API + max_count_param: int = clamp(max_count, 0, 100) # Take range is 0-100 in API - URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={maxCount}&sort=distance+&lite=true" + URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={max_count_param}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually data = await self._request("GET", URL.replace("+", "%2B")) return GeocachingStatus.parse_caches(data) - async def _verify_codes(self, endpoint: str, code_type: str, reference_codes: list[str], extra_params: dict[str, str] = {}) -> None: + async def _verify_codes(self, endpoint: str, code_type: str, reference_codes: set[str], extra_params: dict[str, str] = {}) -> None: """Verifies a set of reference codes to ensure they are valid, and returns a set of all invalid codes""" ref_codes_param: str = ",".join(reference_codes) additional_params: str = "&".join([f'{name}={val}' for name, val in extra_params.items()]) additional_params = "&" + additional_params if len(additional_params) > 0 else "" data = await self._request("GET", f"/{endpoint}?referenceCodes={ref_codes_param}&fields=referenceCode{additional_params}") - invalid_codes: set[str] = set(reference_codes).difference([d["referenceCode"] for d in data]) + invalid_codes: set[str] = reference_codes.difference([d["referenceCode"] for d in data]) if len(invalid_codes) > 0: raise GeocachingInvalidSettingsError(code_type, invalid_codes) diff --git a/geocachingapi/models.py b/geocachingapi/models.py index d42a2ca..75dee54 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -4,6 +4,8 @@ from dataclasses import dataclass, field from datetime import datetime from math import radians, sin, cos, acos + +from geocachingapi.exceptions import GeocachingTooManyCodesError from .utils import try_get_from_dict import reverse_geocode import asyncio @@ -47,21 +49,27 @@ def __init__(self, location: GeocachingCoordinate, radiusKm: float, maxCount: in class GeocachingSettings: """Class to hold the Geocaching API settings""" - tracked_cache_codes: list[str] - tracked_trackable_codes: list[str] + tracked_cache_codes: set[str] + tracked_trackable_codes: set[str] environment: GeocachingApiEnvironment nearby_caches_setting: NearbyCachesSetting - def __init__(self, trackable_codes: list[str] = [], cache_codes: list[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: + def __init__(self, trackable_codes: set[str] = [], cache_codes: set[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" self.tracked_trackable_codes = trackable_codes self.tracked_cache_codes = cache_codes self.nearby_caches_setting = nearby_caches_setting - def set_tracked_caches(self, cache_codes: list[str]): + def set_tracked_caches(self, cache_codes: set[str]): + # A single API call can only fetch 50 caches maximum + if len(cache_codes) > 50: + raise GeocachingTooManyCodesError(f"Number of tracked caches cannot exceed 50. Was: {len(cache_codes)}") self.tracked_cache_codes = cache_codes - def set_tracked_trackables(self, trackable_codes: list[str]): + def set_tracked_trackables(self, trackable_codes: set[str]): + # A single API call can only fetch 50 trackables maximum + if len(trackable_codes) > 50: + raise GeocachingTooManyCodesError(f"Number of tracked trackables cannot exceed 50. Was: {len(trackable_codes)}") self.tracked_trackable_codes = trackable_codes def set_nearby_caches_setting(self, setting: NearbyCachesSetting): diff --git a/geocachingapi/utils.py b/geocachingapi/utils.py index 35b3e39..65fbc4b 100644 --- a/geocachingapi/utils.py +++ b/geocachingapi/utils.py @@ -12,3 +12,7 @@ def try_get_from_dict(data: Dict[str, Any], key: str, fallback: Any, conversion: if conversion is None: return value return conversion(value) + +# Clamps an int between the min and max value, and returns an int in that range +def clamp(value: int, min_value: int, max_value: int) -> int: + return min(max(value, min_value), max_value) \ No newline at end of file From 8171c855058c08033cb32190c22bf7d6107b3083 Mon Sep 17 00:00:00 2001 From: marc7s Date: Sat, 14 Dec 2024 01:17:54 +0100 Subject: [PATCH 37/37] Move limits to limits.py --- geocachingapi/const.py | 2 +- geocachingapi/geocachingapi.py | 3 ++- geocachingapi/limits.py | 5 +++++ geocachingapi/models.py | 9 +++++---- 4 files changed, 13 insertions(+), 6 deletions(-) create mode 100644 geocachingapi/limits.py diff --git a/geocachingapi/const.py b/geocachingapi/const.py index 18334a2..54346e7 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -35,4 +35,4 @@ "userData", "placedDate", "location" - ]) \ No newline at end of file + ]) diff --git a/geocachingapi/geocachingapi.py b/geocachingapi/geocachingapi.py index ec8d45d..81b7e08 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -13,6 +13,7 @@ from typing import Any, Awaitable, Callable, Dict, Optional from .const import ENVIRONMENT_SETTINGS, CACHE_FIELDS_PARAMETER +from .limits import MAXIMUM_NEARBY_CACHES from .exceptions import ( GeocachingApiConnectionError, GeocachingApiConnectionTimeoutError, @@ -293,7 +294,7 @@ async def _update_nearby_caches(self, data: Dict[str, Any] = None) -> None: async def get_nearby_caches(self, coordinates: GeocachingCoordinate, radius_km: float, max_count: int = 10) -> list[GeocachingCache]: """Get caches nearby the provided coordinates, within the provided radius""" radiusM: int = round(radius_km * 1000) # Convert the radius from km to m - max_count_param: int = clamp(max_count, 0, 100) # Take range is 0-100 in API + max_count_param: int = clamp(max_count, 0, MAXIMUM_NEARBY_CACHES) # Take range is 0-100 in API URL = f"/geocaches/search?q=location:[{coordinates.latitude},{coordinates.longitude}]+radius:{radiusM}m&fields={CACHE_FIELDS_PARAMETER}&take={max_count_param}&sort=distance+&lite=true" # The + sign is not encoded correctly, so we encode it manually diff --git a/geocachingapi/limits.py b/geocachingapi/limits.py new file mode 100644 index 0000000..f039206 --- /dev/null +++ b/geocachingapi/limits.py @@ -0,0 +1,5 @@ +"""Geocaching Api Limits.""" + +MAXIMUM_TRACKED_CACHES: int = 50 +MAXIMUM_TRACKED_TRACKABLES: int = 10 +MAXIMUM_NEARBY_CACHES: int = 50 \ No newline at end of file diff --git a/geocachingapi/models.py b/geocachingapi/models.py index 75dee54..b7e9158 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -5,6 +5,7 @@ from datetime import datetime from math import radians, sin, cos, acos +from geocachingapi.limits import MAXIMUM_TRACKED_CACHES, MAXIMUM_TRACKED_TRACKABLES from geocachingapi.exceptions import GeocachingTooManyCodesError from .utils import try_get_from_dict import reverse_geocode @@ -61,14 +62,14 @@ def __init__(self, trackable_codes: set[str] = [], cache_codes: set[str] = [], n self.nearby_caches_setting = nearby_caches_setting def set_tracked_caches(self, cache_codes: set[str]): - # A single API call can only fetch 50 caches maximum - if len(cache_codes) > 50: + # Ensure the number of tracked caches are within the limits + if len(cache_codes) > MAXIMUM_TRACKED_CACHES: raise GeocachingTooManyCodesError(f"Number of tracked caches cannot exceed 50. Was: {len(cache_codes)}") self.tracked_cache_codes = cache_codes def set_tracked_trackables(self, trackable_codes: set[str]): - # A single API call can only fetch 50 trackables maximum - if len(trackable_codes) > 50: + # Ensure the number of tracked trackables are within the limits + if len(trackable_codes) > MAXIMUM_TRACKED_TRACKABLES: raise GeocachingTooManyCodesError(f"Number of tracked trackables cannot exceed 50. Was: {len(trackable_codes)}") self.tracked_trackable_codes = trackable_codes