1"""
  2.. include:: ../README.md
  3"""  # noqa: D200, D212, D400, D415
  4
  5from __future__ import annotations
  6
  7import logging
  8import re
  9from datetime import datetime
 10from typing import Any
 11from uuid import UUID
 12
 13import requests
 14
 15from .const import (
 16    DEFAULT_UUID,
 17    DEXCOM_APPLICATION_IDS,
 18    DEXCOM_AUTHENTICATE_ENDPOINT,
 19    DEXCOM_BASE_URLS,
 20    DEXCOM_GLUCOSE_READINGS_ENDPOINT,
 21    DEXCOM_LOGIN_ID_ENDPOINT,
 22    DEXCOM_TREND_DIRECTIONS,
 23    MAX_MAX_COUNT,
 24    MAX_MINUTES,
 25    MMOL_L_CONVERSION_FACTOR,
 26    TREND_ARROWS,
 27    TREND_DESCRIPTIONS,
 28    Region,
 29)
 30from .errors import (
 31    AccountError,
 32    AccountErrorEnum,
 33    ArgumentError,
 34    ArgumentErrorEnum,
 35    DexcomError,
 36    SessionError,
 37    SessionErrorEnum,
 38)
 39
 40_LOGGER = logging.getLogger("pydexcom")
 41
 42
 43class GlucoseReading:
 44    """Class for parsing glucose reading from Dexcom Share API."""
 45
 46    def __init__(self, json_glucose_reading: dict[str, Any]) -> None:
 47        """Initialize `GlucoseReading` with JSON glucose reading from Dexcom Share API.
 48
 49        :param json_glucose_reading: JSON glucose reading from Dexcom Share API
 50        """
 51        self._json = json_glucose_reading
 52        try:
 53            self._value = int(json_glucose_reading["Value"])
 54            self._trend_direction: str = json_glucose_reading["Trend"]
 55            # Dexcom Share API returns `str` direction now, previously `int` trend
 56            self._trend: int = DEXCOM_TREND_DIRECTIONS[self._trend_direction]
 57
 58            match = re.match(
 59                r"Date\((?P<timestamp>\d+)(?P<timezone>[+-]\d{4})\)",
 60                json_glucose_reading["DT"],
 61            )
 62            if match:
 63                self._datetime = datetime.fromtimestamp(
 64                    int(match.group("timestamp")) / 1000.0,
 65                    tz=datetime.strptime(match.group("timezone"), "%z").tzinfo,
 66                )
 67        except (KeyError, TypeError, ValueError) as error:
 68            raise ArgumentError(ArgumentErrorEnum.GLUCOSE_READING_INVALID) from error
 69
 70    @property
 71    def value(self) -> int:
 72        """Blood glucose value in mg/dL."""
 73        return self._value
 74
 75    @property
 76    def mg_dl(self) -> int:
 77        """Blood glucose value in mg/dL."""
 78        return self._value
 79
 80    @property
 81    def mmol_l(self) -> float:
 82        """Blood glucose value in mmol/L."""
 83        return round(self.value * MMOL_L_CONVERSION_FACTOR, 1)
 84
 85    @property
 86    def trend(self) -> int:
 87        """Blood glucose trend information.
 88
 89        Value of `pydexcom.const.DEXCOM_TREND_DIRECTIONS`.
 90        """
 91        return self._trend
 92
 93    @property
 94    def trend_direction(self) -> str:
 95        """Blood glucose trend direction.
 96
 97        Key of `pydexcom.const.DEXCOM_TREND_DIRECTIONS`.
 98        """
 99        return self._trend_direction
100
101    @property
102    def trend_description(self) -> str | None:
103        """Blood glucose trend information description.
104
105        See `pydexcom.const.TREND_DESCRIPTIONS`.
106        """
107        return TREND_DESCRIPTIONS[self._trend]
108
109    @property
110    def trend_arrow(self) -> str:
111        """Blood glucose trend as unicode arrow (`pydexcom.const.TREND_ARROWS`)."""
112        return TREND_ARROWS[self._trend]
113
114    @property
115    def datetime(self) -> datetime:
116        """Glucose reading recorded time as datetime."""
117        return self._datetime
118
119    @property
120    def json(self) -> dict[str, Any]:
121        """JSON glucose reading from Dexcom Share API."""
122        return self._json
123
124    def __str__(self) -> str:
125        """Blood glucose value as in mg/dL."""
126        return str(self._value)
127
128
129def valid_uuid(uuid: str | None) -> bool:
130    """Check if UUID is valid."""
131    try:
132        UUID(str(uuid))
133    except ValueError:
134        return False
135    else:
136        return True
137
138
139class Dexcom:
140    """Class for communicating with Dexcom Share API."""
141
142    def __init__(
143        self,
144        *,
145        password: str,
146        account_id: str | None = None,
147        username: str | None = None,
148        region: Region = Region.US,
149    ) -> None:
150        """Initialize `Dexcom` with Dexcom Share credentials.
151
152        :param username: username for the Dexcom Share user, *not follower*.
153        :param account_id: account ID for the Dexcom Share user, *not follower*.
154        :param password: password for the Dexcom Share user.
155        :param region: the region to use, one of `"us"`, `"ous"`, `"jp"`.
156        """
157        user_ids = sum(user_id is not None for user_id in [account_id, username])
158        if user_ids == 0:
159            raise ArgumentError(ArgumentErrorEnum.NONE_USER_ID_PROVIDED)
160        if user_ids != 1:
161            raise ArgumentError(ArgumentErrorEnum.TOO_MANY_USER_ID_PROVIDED)
162
163        self._base_url = DEXCOM_BASE_URLS[region]
164        self._application_id = DEXCOM_APPLICATION_IDS[region]
165        self._password = password
166        self._username: str | None = username
167        self._account_id: str | None = account_id
168        self._session_id: str | None = None
169        self.__session = requests.Session()
170        self._session()
171
172    def _post(
173        self,
174        endpoint: str,
175        params: dict[str, Any] | None = None,
176        json: dict[str, Any] | None = None,
177    ) -> Any:  # noqa: ANN401
178        """Send post request to Dexcom Share API.
179
180        :param endpoint: URL of the post request
181        :param params: `dict` to send in the query string of the post request
182        :param json: JSON to send in the body of the post request
183        """
184        response = self.__session.post(
185            f"{self._base_url}/{endpoint}",
186            headers={"Accept-Encoding": "application/json"},
187            params=params,
188            json={} if json is None else json,
189        )
190
191        try:
192            response.raise_for_status()
193            return response.json()
194        except requests.HTTPError as http_error:
195            error = self._handle_response(response)
196            if error:
197                raise error from http_error
198            _LOGGER.exception("%s", response.text)
199            raise
200
201    def _handle_response(self, response: requests.Response) -> DexcomError | None:  # noqa: C901
202        error: DexcomError | None = None
203        """
204        Parse `requests.Response` for `pydexcom.errors.DexcomError`.
205
206        :param response: `requests.Response` to parse
207        """
208        if response.json():
209            _LOGGER.debug("%s", response.json())
210            code = response.json().get("Code", None)
211            message = response.json().get("Message", None)
212            if code == "SessionIdNotFound":
213                error = SessionError(SessionErrorEnum.NOT_FOUND)
214            elif code == "SessionNotValid":
215                error = SessionError(SessionErrorEnum.INVALID)
216            elif code == "AccountPasswordInvalid":  # defunct
217                error = AccountError(AccountErrorEnum.FAILED_AUTHENTICATION)
218            elif code == "SSO_AuthenticateMaxAttemptsExceeded":
219                error = AccountError(AccountErrorEnum.MAX_ATTEMPTS)
220            elif code == "SSO_InternalError":
221                if message and (
222                    "Cannot Authenticate by AccountName" in message
223                    or "Cannot Authenticate by AccountId" in message
224                ):
225                    error = AccountError(AccountErrorEnum.FAILED_AUTHENTICATION)
226            elif code == "InvalidArgument":
227                if message and "accountName" in message:
228                    error = ArgumentError(ArgumentErrorEnum.USERNAME_INVALID)
229                elif message and "password" in message:
230                    error = ArgumentError(ArgumentErrorEnum.PASSWORD_INVALID)
231                elif message and "UUID" in message:
232                    error = ArgumentError(ArgumentErrorEnum.ACCOUNT_ID_INVALID)
233            elif code and message:
234                _LOGGER.debug("%s: %s", code, message)
235        return error
236
237    def _validate_session_id(self) -> None:
238        """Validate session ID."""
239        if any(
240            [
241                not isinstance(self._session_id, str),
242                not self._session_id,
243                not valid_uuid(self._session_id),
244            ],
245        ):
246            raise ArgumentError(ArgumentErrorEnum.SESSION_ID_INVALID)
247        if self._session_id == DEFAULT_UUID:
248            raise ArgumentError(ArgumentErrorEnum.SESSION_ID_DEFAULT)
249
250    def _validate_username(self) -> None:
251        """Validate username."""
252        if any([not isinstance(self._username, str), not self._username]):
253            raise ArgumentError(ArgumentErrorEnum.USERNAME_INVALID)
254
255    def _validate_password(self) -> None:
256        """Validate password."""
257        if any([not isinstance(self._password, str), not self._password]):
258            raise ArgumentError(ArgumentErrorEnum.PASSWORD_INVALID)
259
260    def _validate_account_id(self) -> None:
261        """Validate account ID."""
262        if any(
263            [
264                not isinstance(self._account_id, str),
265                not self._account_id,
266                not valid_uuid(self._account_id),
267            ],
268        ):
269            raise ArgumentError(ArgumentErrorEnum.ACCOUNT_ID_INVALID)
270        if self._account_id == DEFAULT_UUID:
271            raise ArgumentError(ArgumentErrorEnum.ACCOUNT_ID_DEFAULT)
272
273    def _get_account_id(self) -> str:
274        """Retrieve account ID from the authentication endpoint.
275
276        See `pydexcom.const.DEXCOM_AUTHENTICATE_ENDPOINT`.
277        """
278        _LOGGER.debug("Retrieve account ID from the authentication endpoint")
279        return self._post(
280            DEXCOM_AUTHENTICATE_ENDPOINT,
281            json={
282                "accountName": self._username,
283                "password": self._password,
284                "applicationId": self._application_id,
285            },
286        )
287
288    def _get_session_id(self) -> str:
289        """Retrieve session ID from the login endpoint.
290
291        See `pydexcom.const.DEXCOM_LOGIN_ID_ENDPOINT`.
292        """
293        _LOGGER.debug("Retrieve session ID from the login endpoint")
294        return self._post(
295            DEXCOM_LOGIN_ID_ENDPOINT,
296            json={
297                "accountId": self._account_id,
298                "password": self._password,
299                "applicationId": self._application_id,
300            },
301        )
302
303    def _session(self) -> None:
304        """Create Dexcom Share API session."""
305        self._validate_password()
306
307        if self._account_id is None:
308            self._validate_username()
309            self._account_id = self._get_account_id()
310
311        self._validate_account_id()
312        self._session_id = self._get_session_id()
313        self._validate_session_id()
314
315    def _get_glucose_readings(
316        self,
317        minutes: int = MAX_MINUTES,
318        max_count: int = MAX_MAX_COUNT,
319    ) -> list[dict[str, Any]]:
320        """Retrieve glucose readings from the glucose readings endpoint.
321
322        See `pydexcom.const.DEXCOM_GLUCOSE_READINGS_ENDPOINT`.
323        """
324        if not isinstance(minutes, int) or any([minutes < 0, minutes > MAX_MINUTES]):
325            raise ArgumentError(ArgumentErrorEnum.MINUTES_INVALID)
326        if not isinstance(max_count, int) or any(
327            [max_count < 0, max_count > MAX_MAX_COUNT],
328        ):
329            raise ArgumentError(ArgumentErrorEnum.MAX_COUNT_INVALID)
330
331        _LOGGER.debug("Retrieve glucose readings from the glucose readings endpoint")
332        return self._post(
333            DEXCOM_GLUCOSE_READINGS_ENDPOINT,
334            params={
335                "sessionId": self._session_id,
336                "minutes": minutes,
337                "maxCount": max_count,
338            },
339        )
340
341    def get_glucose_readings(
342        self,
343        minutes: int = MAX_MINUTES,
344        max_count: int = MAX_MAX_COUNT,
345    ) -> list[GlucoseReading]:
346        """Get `max_count` glucose readings within specified number of `minutes`.
347
348        Catches one instance of a thrown `pydexcom.errors.SessionError` if session ID
349        expired, attempts to get a new session ID and retries.
350
351        :param minutes: Number of minutes to retrieve glucose readings from (1-1440)
352        :param max_count: Maximum number of glucose readings to retrieve (1-288)
353        """
354        json_glucose_readings: list[dict[str, Any]] = []
355
356        try:
357            # Requesting glucose reading with DEFAULT_UUID returns non-JSON empty string
358            self._validate_session_id()
359
360            json_glucose_readings = self._get_glucose_readings(minutes, max_count)
361        except SessionError:
362            # Attempt to update expired session ID
363            self._session()
364
365            json_glucose_readings = self._get_glucose_readings(minutes, max_count)
366
367        return [GlucoseReading(json_reading) for json_reading in json_glucose_readings]
368
369    def get_latest_glucose_reading(self) -> GlucoseReading | None:
370        """Get latest available glucose reading, within the last 24 hours."""
371        glucose_readings = self.get_glucose_readings(max_count=1)
372        return glucose_readings[0] if glucose_readings else None
373
374    def get_current_glucose_reading(self) -> GlucoseReading | None:
375        """Get current available glucose reading, within the last 10 minutes."""
376        glucose_readings = self.get_glucose_readings(minutes=10, max_count=1)
377        return glucose_readings[0] if glucose_readings else None