diff --git a/examples_sync/challenge_tracker_examples.py b/examples_sync/challenge_tracker_examples.py new file mode 100644 index 0000000..041aa1f --- /dev/null +++ b/examples_sync/challenge_tracker_examples.py @@ -0,0 +1,112 @@ +import os +from getpass import getpass + +from otf_api import OtfSync +from otf_api.models import ChallengeType, EquipmentType + +USERNAME = os.getenv("OTF_EMAIL") or input("Enter your OTF email: ") +PASSWORD = os.getenv("OTF_PASSWORD") or getpass("Enter your OTF password: ") + + +def main(): + with OtfSync(USERNAME, PASSWORD) as otf: + # challenge tracker content is an overview of the challenges OTF runs + # and your participation in them + challenge_tracker_content = otf.get_challenge_tracker_content() + print(challenge_tracker_content.benchmarks[0].model_dump_json(indent=4)) + + """ + { + "equipment_id": 2, + "equipment_name": "Treadmill", + "years": [ + { + "year": "2024", + "is_participated": false, + "in_progress": false + }, + ... + ], + "logo_url": "https://otf-icons.s3.amazonaws.com/benchmarks/Treadmill.png" + } + """ + + print(challenge_tracker_content.challenges[0].model_dump_json(indent=4)) + """ + { + "challenge_category_id": 10, + "challenge_sub_category_id": 8, + "challenge_name": "Catch Me If You Can 3G", + "years": [ + { + "year": "2024", + "is_participated": false, + "in_progress": false + }, + ... + ], + "logo_url": "https://otf-icons.s3.amazonaws.com/challenges/CatchMeIfYouCan.png" + } + """ + + # challenge tracker details are detailed information about specific challenges + # this endpoint takes an equipment type and a challenge type as arguments + tread_challenge_details = otf.get_challenge_tracker_detail(EquipmentType.Treadmill, ChallengeType.Other) + print(tread_challenge_details.details[0].model_dump_json(indent=4)) + + """ + { + "challenge_category_id": 10, + "challenge_sub_category_id": null, + "equipment_id": 2, + "equipment_name": "Treadmill", + "metric_entry": { + "title": "22 MIN", + "equipment_id": 2, + "entry_type": "Distance", + "metric_key": "22MIN", + "min_value": "0.16093440000000003", + "max_value": "8.04672" + }, + "challenge_name": "Catch me If You Can", + "logo_url": "https://otf-icons.s3.amazonaws.com/challenges/CatchMeIfYouCan.png", + "best_record": 1.40012928, + "last_record": 1.40012928, + "previous_record": 1.40012928, + "unit": "km", + "goals": null, + "challenge_histories": [ + { + "challenge_objective": "None", + "challenge_id": 449906, + "studio_id": 1267, + "studio_name": "AnyTown OH - East", + "start_date": "2024-02-06 00:00:00", + "end_date": "2024-02-06 23:59:00", + "total_result": 1.40012928, + "is_finished": true, + "benchmark_histories": [ + { + "studio_name": "AnyTown OH - East", + "equipment_id": 2, + "result": 1.40012928, + "date_created": "2024-02-06 16:01:26", + "date_updated": "2024-02-06 16:01:26", + "class_time": "2024-02-06 09:45:00", + "challenge_sub_category_id": null, + "class_id": 86842386, + "substitute_id": 1, + "weight_lbs": 0, + "workout_type_id": null, + "workout_id": null, + "linked_challenges": [] + } + ] + } + ] + } + """ + + +if __name__ == "__main__": + main() diff --git a/examples_sync/class_bookings_examples.py b/examples_sync/class_bookings_examples.py new file mode 100644 index 0000000..52ed508 --- /dev/null +++ b/examples_sync/class_bookings_examples.py @@ -0,0 +1,124 @@ +import os +from datetime import datetime +from getpass import getpass + +from otf_api import OtfSync +from otf_api.models.classes import DoW + +USERNAME = os.getenv("OTF_EMAIL") or input("Enter your OTF email: ") +PASSWORD = os.getenv("OTF_PASSWORD") or getpass("Enter your OTF password: ") + + +def main(): + with OtfSync(USERNAME, PASSWORD) as otf: + resp = otf.get_bookings(start_date=datetime.today().date()) + print(resp.model_dump_json(indent=4)) + + studios = otf.search_studios_by_geo(40.7831, 73.9712, distance=100) + + studio_uuids = [studio.studio_uuid for studio in studios.studios] + + # To get upcoming classes you can call the `get_classes` method + # You can pass a list of studio_uuids or, if you want to get classes from your home studio, leave it empty + # this also takes a start date, end date, and limit - these are not sent to the API, they are used in the + # client to filter the results + classes = otf.get_classes(studio_uuids, day_of_week=[DoW.TUESDAY, DoW.THURSDAY, DoW.SATURDAY]) + + print(classes.classes[0].model_dump_json(indent=4)) + + """ + { + "id": "0e39ef70-7403-49c1-8605-4a72643bd201", + "ot_base_class_uuid": "08cebfdb-e127-48d4-8a7f-e6ea4dd85c18", + "starts_at": "2024-06-13 10:00:00+00:00", + "starts_at_local": "2024-06-13 05:00:00", + "ends_at": "2024-06-13 11:00:00+00:00", + "ends_at_local": "2024-06-13 06:00:00", + "name": "Orange 3G", + "type": "ORANGE_60", + "studio": ..., + "coach": ..., + "max_capacity": 36, + "booking_capacity": 36, + "waitlist_size": 0, + "full": false, + "waitlist_available": false, + "canceled": false, + "mbo_class_id": "30809", + "mbo_class_schedule_id": "2655", + "mbo_class_description_id": "102", + "created_at": "2024-05-14 10:33:32.406000+00:00", + "updated_at": "2024-06-13 01:58:55.233000+00:00" + } + """ + + # You can also get the classes that you have booked + # You can pass a start_date, end_date, status, and limit as arguments + + bookings = otf.get_bookings() + + print("Latest Upcoming Class:") + print(bookings.bookings[-1].model_dump_json(indent=4)) + + """ + { + "class_booking_id": 870700285, + "class_booking_uuid": "a36d76b1-0a55-4143-b96b-646e7520ca39", + "studio_id": 1234, + "class_id": 376344282, + "is_intro": false, + "member_id": 234488148, + "status": "Booked", + "booked_date": "2024-09-10T04:26:11Z", + "checked_in_date": null, + "cancelled_date": null, + "created_date": "2024-09-10T04:26:11Z", + "updated_date": "2024-09-10T04:26:13Z", + "is_deleted": false, + "waitlist_position": null, + "otf_class": { + "starts_at_local": "2024-09-28T10:30:00", + "ends_at_local": "2024-09-28T11:20:00", + "name": "Tread 50", + "class_uuid": "82ec9b55-950a-484f-818f-cd2344ce83fd", + "is_available": true, + "is_cancelled": false, + "program_name": "Group Fitness", + "coach_id": 1204786, + "studio": { + "studio_uuid": "49e360d1-f8ef-4091-a23f-61b321cb283c", + "studio_name": "AnyTown OH - East", + "description": "", + "status": "Active", + "time_zone": "America/Chicago", + "studio_id": 1267, + "allows_cr_waitlist": true + }, + "coach": { + "coach_uuid": "973516a8-0c6b-41ec-916c-1da9913b9a16", + "name": "Friendly", + "first_name": "Friendly", + "last_name": "Coach" + }, + "location": { + "address_one": "123 S Main St", + "address_two": null, + "city": "AnyTown", + "country": null, + "distance": null, + "latitude": 91.73407745, + "location_name": null, + "longitude": -80.92264626, + "phone_number": "2042348963", + "postal_code": "11111", + "state": "Ohio" + }, + "virtual_class": null + }, + "is_home_studio": true + } + """ + + +if __name__ == "__main__": + main() diff --git a/examples_sync/studio_examples.py b/examples_sync/studio_examples.py new file mode 100644 index 0000000..7a5795b --- /dev/null +++ b/examples_sync/studio_examples.py @@ -0,0 +1,73 @@ +import os +from getpass import getpass + +from otf_api import OtfSync + +USERNAME = os.getenv("OTF_EMAIL") or input("Enter your OTF email: ") +PASSWORD = os.getenv("OTF_PASSWORD") or getpass("Enter your OTF password: ") + + +def main(): + with OtfSync(USERNAME, PASSWORD) as otf: + # if you need to figure out what studios are in an area, you can call `search_studios_by_geo` + # which takes latitude, longitude, distance, page_index, and page_size as arguments + # but you'll generally just need the first 3 + # same as with classes, you can leave it blank and get the studios within 50 miles of your home studio + studios_by_geo = otf.search_studios_by_geo() + print(studios_by_geo.studios[0].model_dump_json(indent=4)) + + """ + { + "studio_id": 1297, + "studio_uuid": "8645fb2b-ef66-4d9d-bda1-f508091ec891", + "mbo_studio_id": 8612481, + "studio_number": "05414", + "studio_name": "AnyTown OH - East", + "studio_physical_location_id": 494, + "time_zone": "America/Chicago", + "contact_email": "studiomanager05414@orangetheoryfitness.com", + "studio_token": "ec2459b2-32b5-4b7e-9759-55270626925a", + "environment": "PROD", + "pricing_level": "", + "tax_rate": "0.000000", + "accepts_visa_master_card": true, + "accepts_american_express": true, + "accepts_discover": true, + "accepts_ach": false, + "is_integrated": true, + "description": "", + "studio_version": "", + "studio_status": "Active", + "open_date": "2017-01-13 00:00:00", + "re_open_date": "2020-05-26 00:00:00", + "studio_type_id": 2, + "sms_package_enabled": false, + "allows_dashboard_access": false, + "allows_cr_waitlist": true, + "cr_waitlist_flag_last_updated": "2020-07-09 02:43:55+00:00", + "royalty_rate": 0, + "marketing_fund_rate": 0, + "commission_percent": 0, + "is_mobile": null, + "is_otbeat": null, + "distance": 0.0, + "studio_location": ..., + "studio_location_localized": ..., + "studio_profiles": { + "is_web": true, + "intro_capacity": 1, + "is_crm": true + }, + "social_media_links": ... + } + """ + + # if you need to get detailed information about a studio, you can call `get_studio_detail` + # which takes a studio_uuid as an argument, but you can leave it blank to get details about your home studio + # this one has a result structure very much like the previous one + studio_detail = otf.get_studio_detail() + print(studio_detail.model_dump_json(indent=4)) + + +if __name__ == "__main__": + main() diff --git a/examples_sync/workout_examples.py b/examples_sync/workout_examples.py new file mode 100644 index 0000000..9a6e58f --- /dev/null +++ b/examples_sync/workout_examples.py @@ -0,0 +1,191 @@ +import os +from getpass import getpass + +from otf_api import OtfSync + +USERNAME = os.getenv("OTF_EMAIL") or input("Enter your OTF email: ") +PASSWORD = os.getenv("OTF_PASSWORD") or getpass("Enter your OTF password: ") + + +def main(): + with OtfSync(USERNAME, PASSWORD) as otf: + resp = otf.get_member_lifetime_stats() + print(resp.model_dump_json(indent=4)) + + resp = otf.get_body_composition_list() + print(resp.data[0].model_dump_json(indent=4)) + + # performance summaries are historical records of your performance in workouts + # `get_performance_summaries` takes a limit (default of 30) and returns a list of summaries + data_list = otf.get_performance_summaries() + print(data_list.summaries[0].model_dump_json(indent=4)) + """ + { + "performance_summary_id": "29dd97f4-3418-4247-b35c-37eabc5e17f3", + "details": { + "calories_burned": 506, + "splat_points": 18, + "step_count": 0, + "active_time_seconds": 3413, + "zone_time_minutes": { + "gray": 2, + "blue": 13, + "green": 24, + "orange": 16, + "red": 2 + } + }, + "ratable": true, + "otf_class": { + "ot_base_class_uuid": "b6549fc2-a479-4b03-9303-e0e45dbcd8c9", + "starts_at_local": "2024-06-11T09:45:00", + "name": "Orange 60 Min 2G", + "coach": ..., + "studio": ..., + }, + "ratings": null + } + """ + + # you can get detailed information about a specific performance summary by calling `get_performance_summary` + # which takes a performance_summary_id as an argument + data = otf.get_performance_summary(data_list.summaries[0].id) + print(data.model_dump_json(indent=4)) + + """ + { + "class_history_uuid": "29dd97f4-3418-4247-b35c-37eabc5e17f3", + "details": { + "calories_burned": 506, + "splat_points": 18, + "step_count": 3314, + "active_time_seconds": 0, + "zone_time_minutes": { + "gray": 2, + "blue": 13, + "green": 24, + "orange": 16, + "red": 2 + }, + "heart_rate": { + "max_hr": 0, + "peak_hr": 180, + "peak_hr_percent": 94, + "avg_hr": 149, + "avg_hr_percent": 78 + }, + "equipment_data": { + "treadmill": { + "avg_pace": { + "display_value": "15:23", + "display_unit": "min/mile", + "metric_value": "923" + }, + "avg_speed": { + "display_value": 3.9, + "display_unit": "mph", + "metric_value": 3.9 + }, + "max_pace": ..., + "max_speed": ..., + "moving_time": ..., + "total_distance": ..., + "avg_incline": ..., + "elevation_gained": ..., + "max_incline": ... + }, + "rower": ... + } + }, + "ratable": false, + "otf_class": { + "starts_at_local": "2024-06-11T09:45:00", + "name": "Orange 60 Min 2G" + } + } + """ + + # telemetry is a detailed record of a specific workout - minute by minute, or more granular if desired + # this endpoint takes a class_history_uuid, as well as a number of max data points - if you do not pass + # this value it will attempt to return enough data points for 30 second intervals + + telemetry = otf.get_telemetry(performance_summary_id=data_list.summaries[0].id) + telemetry.telemetry = telemetry.telemetry[:2] + print(telemetry.model_dump_json(indent=4)) + + """ + { + "member_uuid": "fa323d40-bfae-4e72-872c-e11188d182a7", + "class_history_uuid": "5945a723-930b-449a-bd8f-8267a4ff392f", + "class_start_time": "2024-06-11 14:46:07+00:00", + "max_hr": 191, + "zones": { + "gray": { + "start_bpm": 96, + "end_bpm": 116 + }, + "blue": { + "start_bpm": 117, + "end_bpm": 135 + }, + "green": { + "start_bpm": 136, + "end_bpm": 159 + }, + "orange": { + "start_bpm": 160, + "end_bpm": 175 + }, + "red": { + "start_bpm": 176, + "end_bpm": 191 + } + }, + "window_size": 30, + "telemetry": [ + { + "relative_timestamp": 0, + "hr": 105, + "agg_splats": 0, + "agg_calories": 2, + "timestamp": "2024-06-11 14:46:07+00:00", + "tread_data": { + "tread_speed": 1.34, + "tread_incline": 1.0, + "agg_tread_distance": 9 + }, + "row_data": { + "row_speed": 1.0, + "row_pps": 0.0, + "row_Spm": 0.0, + "agg_row_distance": 0, + "row_pace": 0 + } + }, + { + "relative_timestamp": 30, + "hr": 132, + "agg_splats": 0, + "agg_calories": 4, + "timestamp": "2024-06-11 14:46:37+00:00", + "tread_data": { + "tread_speed": 2.46, + "tread_incline": 1.0, + "agg_tread_distance": 62 + }, + "row_data": { + "row_speed": 1.0, + "row_pps": 0.0, + "row_Spm": 0.0, + "agg_row_distance": 0, + "row_pace": 0 + } + }, + ... + ] + } + """ + + +if __name__ == "__main__": + main() diff --git a/src/otf_api/__init__.py b/src/otf_api/__init__.py index 933ae05..368dfca 100644 --- a/src/otf_api/__init__.py +++ b/src/otf_api/__init__.py @@ -2,6 +2,7 @@ from .api import Otf from .auth import OtfUser +from .sync_api import OtfSync LOG_FMT = "{asctime} - {module}.{funcName}:{lineno} - {levelname} - {message}" DATE_FMT = "%Y-%m-%d %H:%M:%S%z" @@ -11,4 +12,4 @@ __version__ = "0.8.3-dev4" -__all__ = ["Otf", "OtfUser"] +__all__ = ["Otf", "OtfSync", "OtfUser"] diff --git a/src/otf_api/sync_api.py b/src/otf_api/sync_api.py new file mode 100644 index 0000000..d07cdfd --- /dev/null +++ b/src/otf_api/sync_api.py @@ -0,0 +1,949 @@ +import atexit +import contextlib +import json +from datetime import date, datetime, timedelta +from logging import getLogger +from typing import Any + +import httpx +from yarl import URL + +from otf_api import models +from otf_api.auth import OtfUser +from otf_api.exceptions import ( + AlreadyBookedError, + BookingAlreadyCancelledError, + BookingNotFoundError, + OutsideSchedulingWindowError, +) + +API_BASE_URL = "api.orangetheory.co" +API_IO_BASE_URL = "api.orangetheory.io" +API_TELEMETRY_BASE_URL = "api.yuzu.orangetheory.com" +REQUEST_HEADERS = {"Authorization": None, "Content-Type": "application/json", "Accept": "application/json"} +LOGGER = getLogger(__name__) + + +class OtfSync: + user: OtfUser + _session: httpx.Client + + def __init__( + self, + username: str | None = None, + password: str | None = None, + access_token: str | None = None, + id_token: str | None = None, + refresh_token: str | None = None, + device_key: str | None = None, + user: OtfUser | None = None, + ): + """Create a new Otf instance. + + Authentication methods: + --- + - Provide a username and password. + - Provide an access token and id token. + - Provide a user object. + + Args: + username (str, optional): The username of the user. Default is None. + password (str, optional): The password of the user. Default is None. + access_token (str, optional): The access token. Default is None. + id_token (str, optional): The id token. Default is None. + refresh_token (str, optional): The refresh token. Default is None. + device_key (str, optional): The device key. Default is None. + user (OtfUser, optional): A user object. Default is None. + """ + + self.member: models.MemberDetail + self.home_studio_uuid: str + + if user: + self.user = user + elif (username and password) or (access_token and id_token): + self.user = OtfUser( + username=username, + password=password, + access_token=access_token, + id_token=id_token, + refresh_token=refresh_token, + device_key=device_key, + ) + else: + raise ValueError("No valid authentication method provided") + + # simplify access to member_id and member_uuid + self._member_id = self.user.member_id + self._member_uuid = self.user.member_uuid + self._perf_api_headers = { + "koji-member-id": self._member_id, + "koji-member-email": self.user.id_claims_data.email, + } + self.member = self.get_member_detail() + self.home_studio_uuid = self.member.home_studio.studio_uuid + + @property + def headers(self): + """Get the headers for the API request.""" + + # check the token before making a request in case it has expired + + self.user.cognito.check_token() + return { + "Authorization": f"Bearer {self.user.cognito.id_token}", + "Content-Type": "application/json", + "Accept": "application/json", + } + + def __enter__(self): + # Create the session only once when entering the context + self._session = httpx.Client(headers=self.headers) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + # Close the session when exiting the context + if self._session is not None: + self._session.close() + + @property + def session(self): + """Get the httpx session.""" + if not getattr(self, "_session", None): + self._session = httpx.Client(headers=self.headers) + atexit.register(self._close_session) + + return self._session + + def _close_session(self) -> None: + if not hasattr(self, "_session"): + return + + self._session.close() + + def _do( + self, + method: str, + base_url: str, + url: str, + params: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, + **kwargs: Any, + ) -> Any: + """Perform an API request.""" + + params = params or {} + params = {k: v for k, v in params.items() if v is not None} + + full_url = str(URL.build(scheme="https", host=base_url, path=url)) + + LOGGER.debug(f"Making {method!r} request to {full_url}, params: {params}") + + # ensure we have headers that contain the most up-to-date token + if not headers: + headers = self.headers + else: + headers.update(self.headers) + + response = self.session.request(method, full_url, headers=headers, params=params, **kwargs) + + try: + response.raise_for_status() + except httpx.RequestError as e: + LOGGER.exception(f"Error making request: {e}") + LOGGER.exception(f"Response: {response.text}") + except Exception as e: + LOGGER.exception(f"Error making request: {e}") + + return response.json() + + def _classes_request(self, method: str, url: str, params: dict[str, Any] | None = None) -> Any: + """Perform an API request to the classes API.""" + return self._do(method, API_IO_BASE_URL, url, params) + + def _default_request(self, method: str, url: str, params: dict[str, Any] | None = None, **kwargs: Any) -> Any: + """Perform an API request to the default API.""" + return self._do(method, API_BASE_URL, url, params, **kwargs) + + def _telemetry_request(self, method: str, url: str, params: dict[str, Any] | None = None) -> Any: + """Perform an API request to the Telemetry API.""" + return self._do(method, API_TELEMETRY_BASE_URL, url, params) + + def _performance_summary_request( + self, method: str, url: str, headers: dict[str, str], params: dict[str, Any] | None = None + ) -> Any: + """Perform an API request to the performance summary API.""" + return self._do(method, API_IO_BASE_URL, url, params, headers) + + def get_classes( + self, + studio_uuids: list[str] | None = None, + include_home_studio: bool = True, + start_date: str | None = None, + end_date: str | None = None, + limit: int | None = None, + class_type: models.ClassType | list[models.ClassType] | None = None, + exclude_cancelled: bool = False, + day_of_week: list[models.DoW] | None = None, + start_time: list[str] | None = None, + exclude_unbookable: bool = True, + ) -> models.OtfClassList: + """Get the classes for the user. + + Returns a list of classes that are available for the user, based on the studio UUIDs provided. If no studio + UUIDs are provided, it will default to the user's home studio. + + Args: + studio_uuids (list[str] | None): The studio UUIDs to get the classes for. Default is None, which will\ + default to the user's home studio only. + include_home_studio (bool): Whether to include the home studio in the classes. Default is True. + start_date (str | None): The start date to get classes for, in the format "YYYY-MM-DD". Default is None. + end_date (str | None): The end date to get classes for, in the format "YYYY-MM-DD". Default is None. + limit (int | None): Limit the number of classes returned. Default is None. + class_type (ClassType | list[ClassType] | None): The class type to filter by. Default is None. Multiple\ + class types can be provided, if there are multiple there will be a call per class type. + exclude_cancelled (bool): Whether to exclude cancelled classes. Default is False. + day_of_week (list[DoW] | None): The days of the week to filter by. Default is None. + start_time (list[str] | None): The start time to filter by. Default is None. + exclude_unbookable (bool): Whether to exclude classes that are outside the scheduling window. Default is\ + True. + + Returns: + OtfClassList: The classes for the user. + """ + + if not studio_uuids: + studio_uuids = [self.home_studio_uuid] + elif include_home_studio and self.home_studio_uuid not in studio_uuids: + studio_uuids.append(self.home_studio_uuid) + + classes_resp = self._classes_request("GET", "/v1/classes", params={"studio_ids": studio_uuids}) + classes_list = models.OtfClassList(classes=classes_resp["items"]) + + if start_date: + start_dtme = datetime.strptime(start_date, "%Y-%m-%d") # noqa + classes_list.classes = [c for c in classes_list.classes if c.starts_at_local >= start_dtme] + + if end_date: + end_dtme = datetime.strptime(end_date, "%Y-%m-%d") # noqa + classes_list.classes = [c for c in classes_list.classes if c.ends_at_local <= end_dtme] + + if limit: + classes_list.classes = classes_list.classes[:limit] + + if class_type and isinstance(class_type, str): + class_type = [class_type] + + if day_of_week and not isinstance(day_of_week, list): + day_of_week = [day_of_week] + + if start_time and not isinstance(start_time, list): + start_time = [start_time] + + if class_type: + classes_list.classes = [c for c in classes_list.classes if c.class_type in class_type] + + if exclude_cancelled: + classes_list.classes = [c for c in classes_list.classes if not c.canceled] + + for otf_class in classes_list.classes: + otf_class.is_home_studio = otf_class.studio.id == self.home_studio_uuid + + if day_of_week: + classes_list.classes = [c for c in classes_list.classes if c.day_of_week_enum in day_of_week] + + if start_time: + classes_list.classes = [ + c for c in classes_list.classes if any(c.time.strip().startswith(t) for t in start_time) + ] + + classes_list.classes = list(filter(lambda c: not c.canceled, classes_list.classes)) + + if exclude_unbookable: + # this endpoint returns classes that the `book_class` endpoint will reject, this filters them out + max_date = datetime.today().date() + timedelta(days=29) + classes_list.classes = [c for c in classes_list.classes if c.starts_at_local.date() <= max_date] + + booking_resp = self.get_bookings(start_date, end_date, status=models.BookingStatus.Booked) + booked_classes = {b.otf_class.class_uuid for b in booking_resp.bookings} + + for otf_class in classes_list.classes: + otf_class.is_booked = otf_class.ot_class_uuid in booked_classes + + return classes_list + + def get_booking(self, booking_uuid: str) -> models.Booking: + """Get a specific booking by booking_uuid. + + Args: + booking_uuid (str): The booking UUID to get. + + Returns: + BookingList: The booking. + + Raises: + ValueError: If booking_uuid is None or empty string. + """ + if not booking_uuid: + raise ValueError("booking_uuid is required") + + data = self._default_request("GET", f"/member/members/{self._member_id}/bookings/{booking_uuid}") + return models.Booking(**data["data"]) + + def get_booking_by_class(self, class_: str | models.OtfClass) -> models.Booking: + """Get a specific booking by class_uuid or OtfClass object. + + Args: + class_ (str | OtfClass): The class UUID or the OtfClass object to get the booking for. + + Returns: + Booking: The booking. + + Raises: + BookingNotFoundError: If the booking does not exist. + ValueError: If class_uuid is None or empty string. + """ + + class_uuid = class_.ot_class_uuid if isinstance(class_, models.OtfClass) else class_ + + if not class_uuid: + raise ValueError("class_uuid is required") + + all_bookings = self.get_bookings(exclude_cancelled=False, exclude_checkedin=False) + + for booking in all_bookings.bookings: + if booking.otf_class.class_uuid == class_uuid: + return booking + + raise BookingNotFoundError(f"Booking for class {class_uuid} not found.") + + def book_class(self, class_: str | models.OtfClass) -> models.Booking: + """Book a class by providing either the class_uuid or the OtfClass object. + + Args: + class_ (str | OtfClass): The class UUID or the OtfClass object to book. + + Returns: + Booking: The booking. + + Raises: + AlreadyBookedError: If the class is already booked. + OutsideSchedulingWindowError: If the class is outside the scheduling window. + ValueError: If class_uuid is None or empty string. + Exception: If there is an error booking the class. + """ + + class_uuid = class_.ot_class_uuid if isinstance(class_, models.OtfClass) else class_ + if not class_uuid: + raise ValueError("class_uuid is required") + + with contextlib.suppress(BookingNotFoundError): + existing_booking = self.get_booking_by_class(class_uuid) + if existing_booking.status != models.BookingStatus.Cancelled: + raise AlreadyBookedError( + f"Class {class_uuid} is already booked.", booking_uuid=existing_booking.class_booking_uuid + ) + + body = {"classUUId": class_uuid, "confirmed": False, "waitlist": False} + + resp = self._default_request("PUT", f"/member/members/{self._member_id}/bookings", json=body) + + if resp["code"] == "ERROR": + if resp["data"]["errorCode"] == "603": + raise AlreadyBookedError(f"Class {class_uuid} is already booked.") + if resp["data"]["errorCode"] == "602": + raise OutsideSchedulingWindowError(f"Class {class_uuid} is outside the scheduling window.") + + raise Exception(f"Error booking class {class_uuid}: {json.dumps(resp)}") + + # get the booking details - we will only use this to get the booking_uuid + book_class = models.BookClass(**resp["data"]) + + booking = self.get_booking(book_class.booking_uuid) + + return booking + + def cancel_booking(self, booking: str | models.Booking): + """Cancel a booking by providing either the booking_uuid or the Booking object. + + Args: + booking (str | Booking): The booking UUID or the Booking object to cancel. + + Returns: + CancelBooking: The cancelled booking. + + Raises: + ValueError: If booking_uuid is None or empty string + BookingNotFoundError: If the booking does not exist. + """ + booking_uuid = booking.class_booking_uuid if isinstance(booking, models.Booking) else booking + + if not booking_uuid: + raise ValueError("booking_uuid is required") + + try: + self.get_booking(booking_uuid) + except Exception: + raise BookingNotFoundError(f"Booking {booking_uuid} does not exist.") + + params = {"confirmed": "true"} + resp = self._default_request( + "DELETE", f"/member/members/{self._member_id}/bookings/{booking_uuid}", params=params + ) + if resp["code"] == "NOT_AUTHORIZED" and resp["message"].startswith("This class booking has"): + raise BookingAlreadyCancelledError( + f"Booking {booking_uuid} is already cancelled.", booking_uuid=booking_uuid + ) + + return models.CancelBooking(**resp["data"]) + + def get_bookings( + self, + start_date: date | str | None = None, + end_date: date | str | None = None, + status: models.BookingStatus | None = None, + limit: int | None = None, + exclude_cancelled: bool = True, + exclude_checkedin: bool = True, + ): + """Get the member's bookings. + + Args: + start_date (date | str | None): The start date for the bookings. Default is None. + end_date (date | str | None): The end date for the bookings. Default is None. + status (BookingStatus | None): The status of the bookings to get. Default is None, which includes\ + all statuses. Only a single status can be provided. + limit (int | None): The maximum number of bookings to return. Default is None, which returns all\ + bookings. + exclude_cancelled (bool): Whether to exclude cancelled bookings. Default is True. + exclude_checkedin (bool): Whether to exclude checked-in bookings. Default is True. + + Returns: + BookingList: The member's bookings. + + Warning: + --- + Incorrect statuses do not cause any bad status code, they just return no results. + + Tip: + --- + `CheckedIn` - you must provide dates if you want to get bookings with a status of CheckedIn. If you do not + provide dates, the endpoint will return no results for this status. + + Dates Notes: + --- + If dates are provided, the endpoint will return bookings where the class date is within the provided + date range. If no dates are provided, it will go back 45 days and forward about 30 days. + + Developer Notes: + --- + Looking at the code in the app, it appears that this endpoint accepts multiple statuses. Indeed, + it does not throw an error if you include a list of statuses. However, only the last status in the list is + used. I'm not sure if this is a bug or if the API is supposed to work this way. + """ + + if exclude_cancelled and status == models.BookingStatus.Cancelled: + LOGGER.warning( + "Cannot exclude cancelled bookings when status is Cancelled. Setting exclude_cancelled to False." + ) + exclude_cancelled = False + + if isinstance(start_date, date): + start_date = start_date.isoformat() + + if isinstance(end_date, date): + end_date = end_date.isoformat() + + status_value = status.value if status else None + + params = {"startDate": start_date, "endDate": end_date, "statuses": status_value} + + res = self._default_request("GET", f"/member/members/{self._member_id}/bookings", params=params) + + bookings = res["data"][:limit] if limit else res["data"] + + data = models.BookingList(bookings=bookings) + data.bookings = sorted(data.bookings, key=lambda x: x.otf_class.starts_at_local) + + for booking in data.bookings: + if not booking.otf_class: + continue + if booking.otf_class.studio.studio_uuid == self.home_studio_uuid: + booking.is_home_studio = True + else: + booking.is_home_studio = False + + if exclude_cancelled: + data.bookings = [b for b in data.bookings if b.status != models.BookingStatus.Cancelled] + + if exclude_checkedin: + data.bookings = [b for b in data.bookings if b.status != models.BookingStatus.CheckedIn] + + return data + + def _get_bookings_old(self, status: models.BookingStatus | None = None) -> models.BookingList: + """Get the member's bookings. + + Args: + status (BookingStatus | None): The status of the bookings to get. Default is None, which includes + all statuses. Only a single status can be provided. + + Returns: + BookingList: The member's bookings. + + Raises: + ValueError: If an unaccepted status is provided. + + Notes: + --- + This one is called with the param named 'status'. Dates cannot be provided, because if the endpoint + receives a date, it will return as if the param name was 'statuses'. + + Note: This seems to only work for Cancelled, Booked, CheckedIn, and Waitlisted statuses. If you provide + a different status, it will return all bookings, not filtered by status. The results in this scenario do + not line up with the `get_bookings` with no status provided, as that returns fewer records. Likely the + filtered dates are different on the backend. + + My guess: the endpoint called with dates and 'statuses' is a "v2" kind of thing, where they upgraded without + changing the version of the api. Calling it with no dates and a singular (limited) status is probably v1. + + I'm leaving this in here for reference, but marking it private. I just don't want to have to puzzle over + this again if I remove it and forget about it. + + """ + + if status and status not in [ + models.BookingStatus.Cancelled, + models.BookingStatus.Booked, + models.BookingStatus.CheckedIn, + models.BookingStatus.Waitlisted, + ]: + raise ValueError( + "Invalid status provided. Only Cancelled, Booked, CheckedIn, Waitlisted, and None are supported." + ) + + status_value = status.value if status else None + + res = self._default_request( + "GET", f"/member/members/{self._member_id}/bookings", params={"status": status_value} + ) + + return models.BookingList(bookings=res["data"]) + + def get_member_detail( + self, include_addresses: bool = True, include_class_summary: bool = True, include_credit_card: bool = False + ): + """Get the member details. + + Args: + include_addresses (bool): Whether to include the member's addresses in the response. + include_class_summary (bool): Whether to include the member's class summary in the response. + include_credit_card (bool): Whether to include the member's credit card information in the response. + + Returns: + MemberDetail: The member details. + + + Notes: + --- + The include_addresses, include_class_summary, and include_credit_card parameters are optional and determine + what additional information is included in the response. By default, all additional information is included, + with the exception of the credit card information. + + The base member details include the last four of a credit card regardless of the include_credit_card, + although this is not always the same details as what is in the member_credit_card field. There doesn't seem + to be a way to exclude this information, and I do not know which is which or why they differ. + """ + + include: list[str] = [] + if include_addresses: + include.append("memberAddresses") + + if include_class_summary: + include.append("memberClassSummary") + + if include_credit_card: + include.append("memberCreditCard") + + params = {"include": ",".join(include)} if include else None + + data = self._default_request("GET", f"/member/members/{self._member_id}", params=params) + return models.MemberDetail(**data["data"]) + + def get_member_membership(self) -> models.MemberMembership: + """Get the member's membership details. + + Returns: + MemberMembership: The member's membership details. + """ + + data = self._default_request("GET", f"/member/members/{self._member_id}/memberships") + return models.MemberMembership(**data["data"]) + + def get_member_purchases(self) -> models.MemberPurchaseList: + """Get the member's purchases, including monthly subscriptions and class packs. + + Returns: + MemberPurchaseList: The member's purchases. + """ + data = self._default_request("GET", f"/member/members/{self._member_id}/purchases") + return models.MemberPurchaseList(data=data["data"]) + + def get_member_lifetime_stats( + self, select_time: models.StatsTime = models.StatsTime.AllTime + ) -> models.StatsResponse: + """Get the member's lifetime stats. + + Args: + select_time (StatsTime): The time period to get stats for. Default is StatsTime.AllTime. + + Notes: + --- + The time period provided in the path does not do anything, and the endpoint always returns the same data. + It is being provided anyway, in case this changes in the future. + + Returns: + Any: The member's lifetime stats. + """ + + data = self._default_request("GET", f"/performance/v2/{self._member_id}/over-time/{select_time.value}") + + stats = models.StatsResponse(**data["data"]) + return stats + + def get_latest_agreement(self) -> models.LatestAgreement: + """Get the latest agreement for the member. + + Returns: + LatestAgreement: The agreement. + + Notes: + --- + In this context, "latest" means the most recent agreement with a specific ID, not the most recent agreement + in general. The agreement ID is hardcoded in the endpoint, so it will always return the same agreement. + """ + data = self._default_request("GET", "/member/agreements/9d98fb27-0f00-4598-ad08-5b1655a59af6") + return models.LatestAgreement(**data["data"]) + + def get_out_of_studio_workout_history(self) -> models.OutOfStudioWorkoutHistoryList: + """Get the member's out of studio workout history. + + Returns: + OutOfStudioWorkoutHistoryList: The member's out of studio workout history. + """ + data = self._default_request("GET", f"/member/members/{self._member_id}/out-of-studio-workout") + + return models.OutOfStudioWorkoutHistoryList(workouts=data["data"]) + + def get_favorite_studios(self) -> models.FavoriteStudioList: + """Get the member's favorite studios. + + Returns: + FavoriteStudioList: The member's favorite studios. + """ + data = self._default_request("GET", f"/member/members/{self._member_id}/favorite-studios") + + return models.FavoriteStudioList(studios=data["data"]) + + def get_studio_services(self, studio_uuid: str | None = None) -> models.StudioServiceList: + """Get the services available at a specific studio. If no studio UUID is provided, the member's home studio + will be used. + + Args: + studio_uuid (str): The studio UUID to get services for. Default is None, which will use the member's home\ + studio. + + Returns: + StudioServiceList: The services available at the studio. + """ + studio_uuid = studio_uuid or self.home_studio_uuid + data = self._default_request("GET", f"/member/studios/{studio_uuid}/services") + return models.StudioServiceList(data=data["data"]) + + def get_studio_detail(self, studio_uuid: str | None = None) -> models.StudioDetail: + """Get detailed information about a specific studio. If no studio UUID is provided, it will default to the + user's home studio. + + Args: + studio_uuid (str): Studio UUID to get details for. Defaults to None, which will default to the user's home\ + studio. + + Returns: + StudioDetail: Detailed information about the studio. + """ + studio_uuid = studio_uuid or self.home_studio_uuid + + path = f"/mobile/v1/studios/{studio_uuid}" + params = {"include": "locations"} + + res = self._default_request("GET", path, params=params) + return models.StudioDetail(**res["data"]) + + def search_studios_by_geo( + self, + latitude: float | None = None, + longitude: float | None = None, + distance: float = 50, + page_index: int = 1, + page_size: int = 50, + ) -> models.StudioDetailList: + """Search for studios by geographic location. + + Args: + latitude (float, optional): Latitude of the location to search around, if None uses home studio latitude. + longitude (float, optional): Longitude of the location to search around, if None uses home studio longitude. + distance (float, optional): Distance in miles to search around the location. Defaults to 50. + page_index (int, optional): Page index to start at. Defaults to 1. + page_size (int, optional): Number of results per page. Defaults to 50. + + Returns: + StudioDetailList: List of studios that match the search criteria. + + Notes: + --- + There does not seem to be a limit to the number of results that can be requested total or per page, the + library enforces a limit of 50 results per page to avoid potential rate limiting issues. + + """ + path = "/mobile/v1/studios" + + if not latitude and not longitude: + home_studio = self.get_studio_detail() + + latitude = home_studio.studio_location.latitude + longitude = home_studio.studio_location.longitude + + if page_size > 50: + LOGGER.warning("The API does not support more than 50 results per page, limiting to 50.") + page_size = 50 + + if page_index < 1: + LOGGER.warning("Page index must be greater than 0, setting to 1.") + page_index = 1 + + params = { + "pageIndex": page_index, + "pageSize": page_size, + "latitude": latitude, + "longitude": longitude, + "distance": distance, + } + + all_results: list[models.StudioDetail] = [] + + while True: + res = self._default_request("GET", path, params=params) + pagination = models.Pagination(**res["data"].pop("pagination")) + all_results.extend([models.StudioDetail(**studio) for studio in res["data"]["studios"]]) + + if len(all_results) == pagination.total_count: + break + + params["pageIndex"] += 1 + + return models.StudioDetailList(studios=all_results) + + def get_total_classes(self) -> models.TotalClasses: + """Get the member's total classes. This is a simple object reflecting the total number of classes attended, + both in-studio and OT Live. + + Returns: + TotalClasses: The member's total classes. + """ + data = self._default_request("GET", "/mobile/v1/members/classes/summary") + return models.TotalClasses(**data["data"]) + + def get_body_composition_list(self) -> models.BodyCompositionList: + """Get the member's body composition list. + + Returns: + Any: The member's body composition list. + """ + data = self._default_request("GET", f"/member/members/{self._member_uuid}/body-composition") + + return models.BodyCompositionList(data=data["data"]) + + def get_challenge_tracker_content(self) -> models.ChallengeTrackerContent: + """Get the member's challenge tracker content. + + Returns: + ChallengeTrackerContent: The member's challenge tracker content. + """ + data = self._default_request("GET", f"/challenges/v3.1/member/{self._member_id}") + return models.ChallengeTrackerContent(**data["Dto"]) + + def get_challenge_tracker_detail( + self, + equipment_id: models.EquipmentType, + challenge_type_id: models.ChallengeType, + challenge_sub_type_id: int = 0, + ): + """Get the member's challenge tracker details. + + Args: + equipment_id (EquipmentType): The equipment ID. + challenge_type_id (ChallengeType): The challenge type ID. + challenge_sub_type_id (int): The challenge sub type ID. Default is 0. + + Returns: + ChallengeTrackerDetailList: The member's challenge tracker details. + + Notes: + --- + I'm not sure what the challenge_sub_type_id is supposed to be, so it defaults to 0. + + """ + params = { + "equipmentId": equipment_id.value, + "challengeTypeId": challenge_type_id.value, + "challengeSubTypeId": challenge_sub_type_id, + } + + data = self._default_request("GET", f"/challenges/v3/member/{self._member_id}/benchmarks", params=params) + + return models.ChallengeTrackerDetailList(details=data["Dto"]) + + def get_challenge_tracker_participation(self, challenge_type_id: models.ChallengeType) -> Any: + """Get the member's participation in a challenge. + + Args: + challenge_type_id (ChallengeType): The challenge type ID. + + Returns: + Any: The member's participation in the challenge. + + Notes: + --- + I've never gotten this to return anything other than invalid response. I'm not sure if it's a bug + in my code or the API. + + """ + + data = self._default_request( + "GET", + f"/challenges/v1/member/{self._member_id}/participation", + params={"challengeTypeId": challenge_type_id.value}, + ) + return data + + def get_performance_summaries(self, limit: int = 30) -> models.PerformanceSummaryList: + """Get a list of performance summaries for the authenticated user. + + Args: + limit (int): The maximum number of performance summaries to return. Defaults to 30. + + Returns: + PerformanceSummaryList: A list of performance summaries. + + Developer Notes: + --- + In the app, this is referred to as 'getInStudioWorkoutHistory'. + + """ + + res = self._performance_summary_request( + "GET", + "/v1/performance-summaries", + headers=self._perf_api_headers, + params={"limit": limit}, + ) + return models.PerformanceSummaryList(summaries=res["items"]) + + def get_performance_summary(self, performance_summary_id: str) -> models.PerformanceSummaryDetail: + """Get a detailed performance summary for a given workout. + + Args: + performance_summary_id (str): The ID of the performance summary to retrieve. + + Returns: + PerformanceSummaryDetail: A detailed performance summary. + """ + + path = f"/v1/performance-summaries/{performance_summary_id}" + res = self._performance_summary_request("GET", path, headers=self._perf_api_headers) + return models.PerformanceSummaryDetail(**res) + + def get_hr_history(self) -> models.TelemetryHrHistory: + """Get the heartrate history for the user. + + Returns a list of history items that contain the max heartrate, start/end bpm for each zone, + the change from the previous, the change bucket, and the assigned at time. + + Returns: + TelemetryHrHistory: The heartrate history for the user. + + """ + path = "/v1/physVars/maxHr/history" + + params = {"memberUuid": self._member_id} + res = self._telemetry_request("GET", path, params=params) + return models.TelemetryHrHistory(**res) + + def get_max_hr(self) -> models.TelemetryMaxHr: + """Get the max heartrate for the user. + + Returns a simple object that has the member_uuid and the max_hr. + + Returns: + TelemetryMaxHr: The max heartrate for the user. + """ + path = "/v1/physVars/maxHr" + + params = {"memberUuid": self._member_id} + + res = self._telemetry_request("GET", path, params=params) + return models.TelemetryMaxHr(**res) + + def get_telemetry(self, performance_summary_id: str, max_data_points: int = 120) -> models.Telemetry: + """Get the telemetry for a performance summary. + + This returns an object that contains the max heartrate, start/end bpm for each zone, + and a list of telemetry items that contain the heartrate, splat points, calories, and timestamp. + + Args: + performance_summary_id (str): The performance summary id. + max_data_points (int): The max data points to use for the telemetry. Default is 120. + + Returns: + TelemetryItem: The telemetry for the class history. + + """ + path = "/v1/performance/summary" + + params = {"classHistoryUuid": performance_summary_id, "maxDataPoints": max_data_points} + res = self._telemetry_request("GET", path, params=params) + return models.Telemetry(**res) + + # the below do not return any data for me, so I can't test them + + def _get_member_services(self, active_only: bool = True) -> Any: + """Get the member's services. + + Args: + active_only (bool): Whether to only include active services. Default is True. + + Returns: + Any: The member's service + .""" + active_only_str = "true" if active_only else "false" + data = self._default_request( + "GET", f"/member/members/{self._member_id}/services", params={"activeOnly": active_only_str} + ) + return data + + def _get_aspire_data(self, datetime: str | None = None, unit: str | None = None) -> Any: + """Get data from the member's aspire wearable. + + Note: I don't have an aspire wearable, so I can't test this. + + Args: + datetime (str | None): The date and time to get data for. Default is None. + unit (str | None): The measurement unit. Default is None. + + Returns: + Any: The member's aspire data. + """ + params = {"datetime": datetime, "unit": unit} + + data = self._default_request("GET", f"/member/wearables/{self._member_id}/wearable-daily", params=params) + return data