diff --git a/geocachingapi/const.py b/geocachingapi/const.py index cf3f7e2..54346e7 100644 --- a/geocachingapi/const.py +++ b/geocachingapi/const.py @@ -23,3 +23,16 @@ 2: "Charter", 3: "Premium" } + +# Required parameters for fetching caches in order to generate complete GeocachingCache objects +CACHE_FIELDS_PARAMETER: str = ",".join([ + "referenceCode", + "name", + "owner", + "postedCoordinates", + "url", + "favoritePoints", + "userData", + "placedDate", + "location" + ]) diff --git a/geocachingapi/exceptions.py b/geocachingapi/exceptions.py index 79d65a2..104c787 100644 --- a/geocachingapi/exceptions.py +++ b/geocachingapi/exceptions.py @@ -1,8 +1,17 @@ -"""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 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 036f1ac..81b7e08 100644 --- a/geocachingapi/geocachingapi.py +++ b/geocachingapi/geocachingapi.py @@ -11,16 +11,21 @@ from yarl import URL from aiohttp import ClientResponse, ClientSession, ClientError -from typing import Any, Awaitable, Callable, Dict, List, Optional -from .const import ENVIRONMENT_SETTINGS +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, GeocachingApiError, GeocachingApiRateLimitError, + GeocachingInvalidSettingsError, ) +from .utils import clamp from .models import ( + GeocachingCache, + GeocachingCoordinate, GeocachingStatus, GeocachingSettings, GeocachingApiEnvironment, @@ -50,7 +55,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 @@ -90,7 +95,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}", @@ -135,14 +140,35 @@ 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: - await self._update_user(None) - if len(self._settings.trackable_codes) > 0: + # First, update the user + await self._update_user() + + # If we are tracking trackables, update them + if self._tracked_trackables_enabled(): await self._update_trackables() + + # If we are tracking caches, update them + 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 - + async def _update_user(self, data: Dict[str, Any] = None) -> None: assert self._status if data is None: @@ -160,6 +186,15 @@ 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_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 if data is None: @@ -167,6 +202,9 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "referenceCode", "name", "holder", + "owner", + "url", + "releasedDate", "trackingNumber", "kilometersTraveled", "milesTraveled", @@ -175,19 +213,117 @@ async def _update_trackables(self, data: Dict[str, Any] = None) -> None: "isMissing", "type" ]) - trackable_parameters = ",".join(self._settings.trackable_codes) - data = await self._request("GET", f"/trackables?referenceCodes={trackable_parameters}&fields={fields}&expand=trackablelogs:1") + trackable_parameters = ",".join(self._settings.tracked_trackable_codes) + 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 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 + fields = ",".join([ + "referenceCode", + "geocacheName", + "loggedDate", + "coordinates", + "url", + "owner" + ]) + 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 + 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}") + + # 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) + + # 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 + + 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: + """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.") + return + + if data is None: + 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]: + """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, 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 + 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: 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] = 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""" self._settings = settings 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 de46266..b7e9158 100644 --- a/geocachingapi/models.py +++ b/geocachingapi/models.py @@ -1,35 +1,80 @@ from __future__ import annotations -from array import array from enum import Enum from typing import Any, Dict, Optional, TypedDict - -from dataclasses import dataclass +from dataclasses import dataclass, field 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 +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: + """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 - 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: + """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 = max(0, round(maxCount)) class GeocachingSettings: - """Class to hold the Geocaching Api settings""" - trackable_codes: array(str) + """Class to hold the Geocaching API settings""" + tracked_cache_codes: set[str] + tracked_trackable_codes: set[str] environment: GeocachingApiEnvironment + nearby_caches_setting: NearbyCachesSetting - def __init__(self, environment:GeocachingApiEnvironment = GeocachingApiEnvironment.Production, trackables:array(str) = [] ) -> None: + def __init__(self, trackable_codes: set[str] = [], cache_codes: set[str] = [], nearby_caches_setting: NearbyCachesSetting = None) -> None: """Initialize settings""" - self.trackable_codes = trackables - - def set_trackables(self, trackables:array(str)): - self.trackable_codes = trackables + 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: set[str]): + # 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]): + # 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 + + def set_nearby_caches_setting(self, setting: NearbyCachesSetting): + self.nearby_caches_setting = setting @dataclass class GeocachingUser: @@ -65,19 +110,58 @@ 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. 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 - logged_date: Optional[datetime] = None + 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 - def __init__(self, *, data: Dict[str, Any]) -> GeocachingTrackableJourney: + # 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: + 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 + async def from_list(cls, data_list: list[Dict[str, Any]]) -> list[GeocachingTrackableJourney]: + """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 @dataclass class GeocachingTrackableLog: @@ -88,16 +172,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 @@ -106,60 +185,138 @@ class GeocachingTrackable: reference_code: Optional[str] = None 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[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 - is_missing: bool = False, - trackable_type: str = None, + journeys: Optional[list[GeocachingTrackableJourney]] = field(default_factory=list) + coordinates: GeocachingCoordinate = 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.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.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) - self.miles_traveled = try_get_from_dict(data, "milesTraveled", self.miles_traveled) - self.current_geocache_code = try_get_from_dict(data, "currectGeocacheCode", self.current_geocache_code) + 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, "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) + 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 + 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 + 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.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) + + # 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.found_date_time = found_date_time + self.found_by_user = found_date_time is not None + else: + self.found_date_time = None + self.found_by_user = None + + # 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") + # 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: + 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] = [] + tracked_caches: list[GeocachingCache] = [] def __init__(self): """Initialize GeocachingStatus""" self.user = GeocachingUser() self.trackables = {} + self.nearby_caches = [] + self.tracked_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 + + self.tracked_caches = GeocachingStatus.parse_caches(data) + 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(): self.trackables[reference_code] = GeocachingTrackable() self.trackables[reference_code].update_from_dict(trackable) - \ No newline at end of file + + def update_nearby_caches_from_dict(self, data: Any) -> None: + """Update nearby caches from the API result""" + if not any(data): + pass + + 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) + caches.append(cache) + + return caches \ No newline at end of file diff --git a/geocachingapi/utils.py b/geocachingapi/utils.py index a896666..65fbc4b 100644 --- a/geocachingapi/utils.py +++ b/geocachingapi/utils.py @@ -1,14 +1,18 @@ """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 None + 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) + +# 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 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", diff --git a/tests/test.py b/tests/test.py index 20f930e..dee7225 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() @@ -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}')