camply

camply init file

AvailableCampsite

Bases: CamplyModel

Campsite Storage

This container should be universal regardless of API Provider

Source code in camply/containers/data_containers.py

class AvailableCampsite(CamplyModel):
    """
    Campsite Storage

    This container should be universal regardless of API Provider
    """

    campsite_id: Union[int, str]
    booking_date: datetime.datetime
    booking_end_date: datetime.datetime
    booking_nights: int
    campsite_site_name: str
    campsite_loop_name: Optional[str]
    campsite_type: Optional[str]
    campsite_occupancy: Tuple[int, int]
    campsite_use_type: Optional[str]
    availability_status: str
    recreation_area: str
    recreation_area_id: Union[int, str]
    facility_name: str
    facility_id: Union[int, str]
    booking_url: str
    location: Optional[CampsiteLocation] = None

    permitted_equipment: Optional[List[RecDotGovEquipment]]
    campsite_attributes: Optional[List[RecDotGovAttribute]]

    __unhashable__ = {"permitted_equipment", "campsite_attributes", "location"}

EquipmentOptions

Bases: str, Enum

Enumeration of the Equipment Options

Source code in camply/config/search_config.py

class EquipmentOptions(str, Enum):
    """
    Enumeration of the Equipment Options
    """

    tent = "tent"
    rv = "rv"
    trailer = "trailer"
    vehicle = "vehicle"
    other = "other"

    __all_accepted_equipment__ = [tent, rv, trailer, vehicle]

GoingToCamp

Bases: BaseProvider

Going To Camp API provider

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

class GoingToCamp(BaseProvider):
    """
    Going To Camp API provider
    """

    @classmethod
    def find_recreation_areas(
        cls, search_string: Optional[str] = None, **kwargs
    ) -> List[RecreationArea]:
        """
        Find Matching Recreation Areas based on search string

        Parameters
        ----------
        search_string: Optional[str]
            Search Keyword(s)

        Returns
        -------
        filtered_responses: List[RecreationArea]
            Array of Matching Recreation Areas
        """
        if search_string is not None:
            logger.info(f'Searching for Recreation Areas matching: "{search_string}"')

        if not search_string or search_string == "":
            rec_areas = RECREATION_AREAS.values()
            log_sorted_response(rec_areas)
            return rec_areas

        rec_areas = []
        for _, rec_area in RECREATION_AREAS.items():
            if (
                search_string.lower() in rec_area.recreation_area.lower()
                or search_string.lower() in rec_area.recreation_area_location.lower()
            ):
                rec_areas.append(rec_area)

        log_sorted_response(rec_areas)

        return rec_areas

    def rec_area_lookup(self, rec_area_id: int) -> Tuple[str, RecreationArea]:
        """
        Lookup a recreation area by ID

        Parameters
        ----------
        rec_area_id: int
            The recreation area ID to lookup

        Returns
        -------
        domain_name, rec_ara: Tuple[str, RecreationArea]
            The rec area's domain name and the recreation area object
        """
        for domain_name, rec_area in RECREATION_AREAS.items():
            if str(rec_area.recreation_area_id) == str(rec_area_id):
                return domain_name, rec_area

    def find_campgrounds(
        self,
        search_string: Optional[str] = None,
        rec_area_id: Optional[List[int]] = None,
        campground_id: Optional[List[int]] = None,
        campsite_id: Optional[List[int]] = None,
        **kwargs,
    ) -> List[CampgroundFacility]:
        """
        Find Campgrounds Given a Set of Search Criteria

        Parameters
        ----------
        search_string: Optional[str]
            Search Keyword(s)
        rec_area_id: Optional[List[int]]
            Recreation Area ID by which to filter
        campground_id: Optional[List[int]]
            ID of the Campground

        Returns
        -------
        facilities: List[CampgroundFacility]
            Array of Matching Campgrounds
        """
        if rec_area_id in (None, [], ()):
            logger.error(
                "This provider requires --rec-area to be specified when seaching for campsites"
            )
            sys.exit(1)

        return self.find_facilities_per_recreation_area(
            rec_area_id=rec_area_id,
            campground_id=campground_id,
            search_string=search_string,
        )

    def _get_attr_val(self, attribute, attribute_detail) -> any:
        for attr_value in attribute.get("values", []):
            for attribute_enum_detail in attribute_detail.get("values"):
                if attribute_enum_detail["enumValue"] == attr_value:
                    return _fetch_nested_key(
                        attribute_enum_detail, "localizedValues", 0, "displayName"
                    )

    def get_site_details(self, rec_area_id: int, resource_id: int):
        """
        Get the details about a site in a recreation area

        Parameters
        ----------
        rec_area_id: int
            Recreation Area ID by which to filter
        resource_id: int

        Returns
        -------
        details: Dict[str, str]
            The details about the site
        """
        if not hasattr(self, "_attribute_details"):
            self._attribute_details = self._api_request(
                rec_area_id, "ATTRIBUTE_DETAILS"
            )
        attribute_details = self._attribute_details

        site_details = self._api_request(
            rec_area_id, "SITE_DETAILS", {"resourceId": resource_id}
        )
        site_attributes = {}
        for attribute in site_details["definedAttributes"]:
            attribute_detail = attribute_details[
                f"{attribute['attributeDefinitionId']}"
            ]
            attribute_name = _fetch_nested_key(
                attribute_detail, "localizedValues", 0, "displayName"
            )
            attribute_value = attribute.get("value")
            attribute_values = []
            # Attribute a multi-value enum
            if not attribute_value:
                attr_value = self._get_attr_val(attribute, attribute_detail)
                if not attr_value:
                    continue
                attribute_values.append(attr_value)
            else:
                attribute_values.append(f"{attribute_value}")

            site_attributes[attribute_name] = ",".join(attribute_values)
        site_details["site_attributes"] = site_attributes

        return site_details

    def get_reservation_link(
        self,
        rec_area_domain_name,
        resource_location_id,
        map_id,
        equipment_id,
        sub_equipment_id,
        party_size,
        start_date,
        end_date,
    ):
        """
        Generate a URL which a site can be booked

        Returns
        -------
        url: str
            The reservation link URL

        """
        if not sub_equipment_id:
            sub_equipment_id = ""

        return (
            "https://%s/create-booking/results?mapId=%s"
            "&bookingCategoryId=0"
            "&startDate=%s"
            "&endDate=%s"
            "&isReserving=true"
            "&equipmentId=%s"
            "&subEquipmentId=%s"
            "&partySize=%s"
            "&resourceLocationId=%s"
            % (
                rec_area_domain_name,
                map_id,
                start_date.isoformat(),
                end_date.isoformat(),
                equipment_id,
                sub_equipment_id,
                party_size,
                resource_location_id,
            )
        )

    def find_facilities_per_recreation_area(
        self,
        rec_area_id: Optional[Union[List[int], int]] = None,
        campground_id: Optional[Union[List[int], int]] = None,
        search_string: Optional[str] = None,
        **kwargs,
    ) -> List[CampgroundFacility]:
        """
        Find Matching Campsites by Recreation Area

        Parameters
        ----------
        rec_area_id: Optional[Union[List[int], int]]
            Recreation Area ID
        campground_id: Optional[Union[List[int], int]]
            Campground IDs
        search_string: Optional[str]
            A string to search for in the facility name

        Returns
        -------
        campgrounds: List[CampgroundFacility]
            Array of Matching Campsites
        """
        rec_area_id = make_list(rec_area_id, coerce=int)[0]
        logger.info(
            f"Retrieving Facility Information for Recreation Area ID: `{rec_area_id}`."
        )

        rec_area = None
        for _, ra in RECREATION_AREAS.items():
            if str(ra.recreation_area_id) == str(rec_area_id):
                rec_area = ra
        if not rec_area:
            logger.error(f"Recreation area '{rec_area_id}' does not exist.")
            sys.exit(1)

        self.campground_details = {}
        api_response = self._api_request(rec_area_id, "LIST_CAMPGROUNDS")

        filtered_facilities = self._filter_facilities_responses(
            rec_area_id, facilities=api_response
        )

        campgrounds = []
        # Fetch campgrounds details for all facilities
        for camp_details in self._api_request(rec_area_id, "CAMP_DETAILS"):
            self.campground_details[camp_details["resourceLocationId"]] = camp_details

        # If a search string is provided, make sure every facility name contains
        # the search string
        if search_string and search_string not in [[], (), ""]:
            filtered_facilities = [
                f
                for f in filtered_facilities
                if search_string.lower() in f.resource_location_name.lower()
            ]

        for facility in filtered_facilities:
            _, campground_facility = self._process_facilities_responses(
                rec_area, facility=facility
            )
            if not campground_facility:
                continue
            if not campground_id:
                campgrounds.append(campground_facility)
            campground_strings = make_list(campground_id, coerce=str)
            if (
                campground_id
                and str(campground_facility.facility_id) in campground_strings
            ):
                campgrounds.append(campground_facility)
        logger.info(f"{len(campgrounds)} Matching Campgrounds Found")
        log_sorted_response(response_array=campgrounds)
        return campgrounds

    def _hostname_for(self, recreation_area_id: int) -> str:
        for hostname, recreation_area in RECREATION_AREAS.items():
            if str(recreation_area.recreation_area_id) == str(recreation_area_id):
                return hostname
        return None

    def _api_request(
        self,
        rec_area_id: int,
        endpoint_name: str,
        params: Optional[Dict[str, str]] = None,
    ) -> str:
        if params is None:
            params = {}

        hostname = self._hostname_for(rec_area_id)
        endpoint = ENDPOINTS.get(endpoint_name)
        url = None
        if endpoint:
            url = endpoint.format(hostname)
        headers = {
            "User-Agent": UserAgent(browsers=["chrome"]).random,
            "Accept-Language": "en-US,en;q=0.9",
        }
        response = requests.get(url=url, headers=headers, params=params, timeout=30)
        if response.ok is False:
            error_message = f"Receiving bad data from GoingToCamp API: status_code: {response.status_code}: {response.text}"
            logger.error(error_message)
            raise ConnectionError(error_message)

        return json.loads(response.content)

    def _filter_facilities_responses(
        self, rec_area_id: int, facilities=List[Dict[str, Any]]
    ) -> List[ResourceLocation]:
        """
        Filter Facilities to Actual Reservable Campsites

        Parameters
        ----------
        rec_area_id: int
            Recreation Area ID
        facilities: List[Dict[str, Any]]
            List of facilities

        Returns
        -------
        List[ResourceLocation]
        """
        filtered_facilities = []
        for facil in facilities:
            try:
                location_name = _fetch_nested_key(
                    facil, "localizedValues", 0, "fullName"
                )
                park_alerts = _fetch_nested_key(
                    facil, "park_alerts", "en-US", 0, "messageTitle"
                )
                if not park_alerts:
                    park_alerts = _fetch_nested_key(
                        facil, "park_alerts", "en-CA", 0, "messageTitle"
                    )

                region_name = _fetch_nested_key(facil, "region")

                facility = ResourceLocation(
                    id=None,
                    region_name=region_name if region_name else "",
                    park_alerts=park_alerts,
                    rec_area_id=rec_area_id,
                    resource_categories=facil.get("resourceCategoryIds"),
                    resource_location_id=facil.get("resourceLocationId"),
                    resource_location_name=location_name,
                )
            except ValidationError as ve:
                logger.error("That doesn't look like a valid Campground Facility")
                logger.error(facil)
                raise ProviderSearchError(
                    "Invalid Campground Facility Returned"
                ) from ve

            if not facility.resource_categories:
                continue

            # Resource categories from: /api/resourcecategory
            if any(
                [
                    CAMP_SITE in facility.resource_categories,
                    GROUP_SITE in facility.resource_categories,
                    OVERFLOW_SITE in facility.resource_categories,
                ]
            ):
                filtered_facilities.append(facility)

        return filtered_facilities

    def _process_facilities_responses(
        self, rec_area: RecreationArea, facility: ResourceLocation
    ) -> Tuple[dict, Optional[CampgroundFacility]]:
        """
        Process Facilities Responses to be More Usable

        Parameters
        ----------
        facility: dict

        Returns
        -------
        Tuple[dict, CampgroundFacility]
        """
        self.campground_details[facility.resource_location_id]
        facility.id = _fetch_nested_key(
            self.campground_details, facility.resource_location_id, "mapId"
        )
        if facility.region_name:
            formatted_recreation_area = (
                f"{rec_area.recreation_area}, {facility.region_name}"
            )
        else:
            formatted_recreation_area = f"{rec_area.recreation_area}"

        campground_facility = CampgroundFacility(
            facility_name=facility.resource_location_name,
            recreation_area=formatted_recreation_area,
            facility_id=facility.resource_location_id,
            recreation_area_id=facility.rec_area_id,
            map_id=facility.id,
        )
        return facility, campground_facility

    def _find_matching_resources(self, rec_area_id: int, search_filter: Dict[str, any]):
        results = self._api_request(rec_area_id, "MAPDATA", search_filter)
        availability_details = {
            search_filter["mapId"]: results["resourceAvailabilities"]
        }

        return availability_details, list(results["mapLinkAvailabilities"].keys())

    def list_equipment_types(self, rec_area_id: int) -> Dict[str, int]:
        """
        List equipment types available for a recreation area

        Params
        ------
        rec_area_id: int
            The ID of the recreation area

        Returns
        -------
        types: List[GoingToCampEquipment]
            A list of equipment types available to this rec area
        """
        results = self._api_request(rec_area_id, "LIST_EQUIPMENT")

        equipment_types = []
        # Only allow equipment from non-group equipment category (the 0th
        # element in results)
        for sub_category in results[0]["subEquipmentCategories"]:
            equipment_name = _fetch_nested_key(
                sub_category, "localizedValues", 0, "name"
            )
            equipment_id = sub_category["subEquipmentCategoryId"]
            equipment_types.append(
                GoingToCampEquipment(
                    equipment_name=equipment_name, equipment_type_id=equipment_id
                )
            )

        log_sorted_response(response_array=equipment_types)
        return equipment_types

    def list_site_availability(
        self,
        campground: CampgroundFacility,
        start_date: datetime.date,
        end_date: datetime.date,
        equipment_type_id: Optional[str],
    ) -> List[AvailableResource]:
        """
        Retrieve the Availability for all Sites in a Camp Area

        Sites are filtered on the provided date range and compatible
        equipment.

        Returns
        -------
        available_sites: List[AvailableResource]
            The list of available sites
        """
        search_filter = {
            "mapId": campground.map_id,
            "resourceLocationId": campground.facility_id,
            "bookingCategoryId": 0,
            "startDate": start_date.isoformat(),
            "endDate": end_date.isoformat(),
            "isReserving": True,
            "getDailyAvailability": False,
            "partySize": 1,
            "numEquipment": 1,
            "equipmentCategoryId": NON_GROUP_EQUIPMENT,
            "filterData": [],
        }
        if equipment_type_id:
            search_filter["subEquipmentCategoryId"] = equipment_type_id

        resources, additional_resources = self._find_matching_resources(
            campground.recreation_area_id, search_filter
        )

        # Resources are often deeply nested; fetch nested resources
        for map_id in additional_resources:
            search_filter["mapId"] = map_id
            avail, _ = self._find_matching_resources(
                campground.recreation_area_id, search_filter
            )
            resources.update(avail)

        availabilities = []
        for map_id, resource_details in resources.items():
            for resource_id, availability_details in resource_details.items():
                if availability_details[0]["availability"] == 0:
                    ar = AvailableResource(resource_id=resource_id, map_id=map_id)
                    availabilities.append(ar)

        return availabilities

find_campgrounds(search_string=None, rec_area_id=None, campground_id=None, campsite_id=None, **kwargs)

Find Campgrounds Given a Set of Search Criteria

Parameters:

Name	Type	Description	Default
search_string	Optional[str]	Search Keyword(s)	None
rec_area_id	Optional[List[int]]	Recreation Area ID by which to filter	None
campground_id	Optional[List[int]]	ID of the Campground	None

Returns:

Name	Type	Description
facilities	List[CampgroundFacility]	Array of Matching Campgrounds

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def find_campgrounds(
    self,
    search_string: Optional[str] = None,
    rec_area_id: Optional[List[int]] = None,
    campground_id: Optional[List[int]] = None,
    campsite_id: Optional[List[int]] = None,
    **kwargs,
) -> List[CampgroundFacility]:
    """
    Find Campgrounds Given a Set of Search Criteria

    Parameters
    ----------
    search_string: Optional[str]
        Search Keyword(s)
    rec_area_id: Optional[List[int]]
        Recreation Area ID by which to filter
    campground_id: Optional[List[int]]
        ID of the Campground

    Returns
    -------
    facilities: List[CampgroundFacility]
        Array of Matching Campgrounds
    """
    if rec_area_id in (None, [], ()):
        logger.error(
            "This provider requires --rec-area to be specified when seaching for campsites"
        )
        sys.exit(1)

    return self.find_facilities_per_recreation_area(
        rec_area_id=rec_area_id,
        campground_id=campground_id,
        search_string=search_string,
    )

find_facilities_per_recreation_area(rec_area_id=None, campground_id=None, search_string=None, **kwargs)

Find Matching Campsites by Recreation Area

Parameters:

Name	Type	Description	Default
rec_area_id	Optional[Union[List[int], int]]	Recreation Area ID	None
campground_id	Optional[Union[List[int], int]]	Campground IDs	None
search_string	Optional[str]	A string to search for in the facility name	None

Returns:

Name	Type	Description
campgrounds	List[CampgroundFacility]	Array of Matching Campsites

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def find_facilities_per_recreation_area(
    self,
    rec_area_id: Optional[Union[List[int], int]] = None,
    campground_id: Optional[Union[List[int], int]] = None,
    search_string: Optional[str] = None,
    **kwargs,
) -> List[CampgroundFacility]:
    """
    Find Matching Campsites by Recreation Area

    Parameters
    ----------
    rec_area_id: Optional[Union[List[int], int]]
        Recreation Area ID
    campground_id: Optional[Union[List[int], int]]
        Campground IDs
    search_string: Optional[str]
        A string to search for in the facility name

    Returns
    -------
    campgrounds: List[CampgroundFacility]
        Array of Matching Campsites
    """
    rec_area_id = make_list(rec_area_id, coerce=int)[0]
    logger.info(
        f"Retrieving Facility Information for Recreation Area ID: `{rec_area_id}`."
    )

    rec_area = None
    for _, ra in RECREATION_AREAS.items():
        if str(ra.recreation_area_id) == str(rec_area_id):
            rec_area = ra
    if not rec_area:
        logger.error(f"Recreation area '{rec_area_id}' does not exist.")
        sys.exit(1)

    self.campground_details = {}
    api_response = self._api_request(rec_area_id, "LIST_CAMPGROUNDS")

    filtered_facilities = self._filter_facilities_responses(
        rec_area_id, facilities=api_response
    )

    campgrounds = []
    # Fetch campgrounds details for all facilities
    for camp_details in self._api_request(rec_area_id, "CAMP_DETAILS"):
        self.campground_details[camp_details["resourceLocationId"]] = camp_details

    # If a search string is provided, make sure every facility name contains
    # the search string
    if search_string and search_string not in [[], (), ""]:
        filtered_facilities = [
            f
            for f in filtered_facilities
            if search_string.lower() in f.resource_location_name.lower()
        ]

    for facility in filtered_facilities:
        _, campground_facility = self._process_facilities_responses(
            rec_area, facility=facility
        )
        if not campground_facility:
            continue
        if not campground_id:
            campgrounds.append(campground_facility)
        campground_strings = make_list(campground_id, coerce=str)
        if (
            campground_id
            and str(campground_facility.facility_id) in campground_strings
        ):
            campgrounds.append(campground_facility)
    logger.info(f"{len(campgrounds)} Matching Campgrounds Found")
    log_sorted_response(response_array=campgrounds)
    return campgrounds

find_recreation_areas(search_string=None, **kwargs) classmethod

Find Matching Recreation Areas based on search string

Parameters:

Name	Type	Description	Default
search_string	Optional[str]	Search Keyword(s)	None

Returns:

Name	Type	Description
filtered_responses	List[RecreationArea]	Array of Matching Recreation Areas

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

@classmethod
def find_recreation_areas(
    cls, search_string: Optional[str] = None, **kwargs
) -> List[RecreationArea]:
    """
    Find Matching Recreation Areas based on search string

    Parameters
    ----------
    search_string: Optional[str]
        Search Keyword(s)

    Returns
    -------
    filtered_responses: List[RecreationArea]
        Array of Matching Recreation Areas
    """
    if search_string is not None:
        logger.info(f'Searching for Recreation Areas matching: "{search_string}"')

    if not search_string or search_string == "":
        rec_areas = RECREATION_AREAS.values()
        log_sorted_response(rec_areas)
        return rec_areas

    rec_areas = []
    for _, rec_area in RECREATION_AREAS.items():
        if (
            search_string.lower() in rec_area.recreation_area.lower()
            or search_string.lower() in rec_area.recreation_area_location.lower()
        ):
            rec_areas.append(rec_area)

    log_sorted_response(rec_areas)

    return rec_areas

get_reservation_link(rec_area_domain_name, resource_location_id, map_id, equipment_id, sub_equipment_id, party_size, start_date, end_date)

Generate a URL which a site can be booked

Returns:

Name	Type	Description
url	str	The reservation link URL

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def get_reservation_link(
    self,
    rec_area_domain_name,
    resource_location_id,
    map_id,
    equipment_id,
    sub_equipment_id,
    party_size,
    start_date,
    end_date,
):
    """
    Generate a URL which a site can be booked

    Returns
    -------
    url: str
        The reservation link URL

    """
    if not sub_equipment_id:
        sub_equipment_id = ""

    return (
        "https://%s/create-booking/results?mapId=%s"
        "&bookingCategoryId=0"
        "&startDate=%s"
        "&endDate=%s"
        "&isReserving=true"
        "&equipmentId=%s"
        "&subEquipmentId=%s"
        "&partySize=%s"
        "&resourceLocationId=%s"
        % (
            rec_area_domain_name,
            map_id,
            start_date.isoformat(),
            end_date.isoformat(),
            equipment_id,
            sub_equipment_id,
            party_size,
            resource_location_id,
        )
    )

get_site_details(rec_area_id, resource_id)

Get the details about a site in a recreation area

Parameters:

Name	Type	Description	Default
rec_area_id	int	Recreation Area ID by which to filter	required
resource_id	int		required

Returns:

Name	Type	Description
details	Dict[str, str]	The details about the site

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def get_site_details(self, rec_area_id: int, resource_id: int):
    """
    Get the details about a site in a recreation area

    Parameters
    ----------
    rec_area_id: int
        Recreation Area ID by which to filter
    resource_id: int

    Returns
    -------
    details: Dict[str, str]
        The details about the site
    """
    if not hasattr(self, "_attribute_details"):
        self._attribute_details = self._api_request(
            rec_area_id, "ATTRIBUTE_DETAILS"
        )
    attribute_details = self._attribute_details

    site_details = self._api_request(
        rec_area_id, "SITE_DETAILS", {"resourceId": resource_id}
    )
    site_attributes = {}
    for attribute in site_details["definedAttributes"]:
        attribute_detail = attribute_details[
            f"{attribute['attributeDefinitionId']}"
        ]
        attribute_name = _fetch_nested_key(
            attribute_detail, "localizedValues", 0, "displayName"
        )
        attribute_value = attribute.get("value")
        attribute_values = []
        # Attribute a multi-value enum
        if not attribute_value:
            attr_value = self._get_attr_val(attribute, attribute_detail)
            if not attr_value:
                continue
            attribute_values.append(attr_value)
        else:
            attribute_values.append(f"{attribute_value}")

        site_attributes[attribute_name] = ",".join(attribute_values)
    site_details["site_attributes"] = site_attributes

    return site_details

list_equipment_types(rec_area_id)

List equipment types available for a recreation area

Params

rec_area_id: int The ID of the recreation area

Returns:

Name	Type	Description
types	List[GoingToCampEquipment]	A list of equipment types available to this rec area

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def list_equipment_types(self, rec_area_id: int) -> Dict[str, int]:
    """
    List equipment types available for a recreation area

    Params
    ------
    rec_area_id: int
        The ID of the recreation area

    Returns
    -------
    types: List[GoingToCampEquipment]
        A list of equipment types available to this rec area
    """
    results = self._api_request(rec_area_id, "LIST_EQUIPMENT")

    equipment_types = []
    # Only allow equipment from non-group equipment category (the 0th
    # element in results)
    for sub_category in results[0]["subEquipmentCategories"]:
        equipment_name = _fetch_nested_key(
            sub_category, "localizedValues", 0, "name"
        )
        equipment_id = sub_category["subEquipmentCategoryId"]
        equipment_types.append(
            GoingToCampEquipment(
                equipment_name=equipment_name, equipment_type_id=equipment_id
            )
        )

    log_sorted_response(response_array=equipment_types)
    return equipment_types

list_site_availability(campground, start_date, end_date, equipment_type_id)

Retrieve the Availability for all Sites in a Camp Area

Sites are filtered on the provided date range and compatible equipment.

Returns:

Name	Type	Description
available_sites	List[AvailableResource]	The list of available sites

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def list_site_availability(
    self,
    campground: CampgroundFacility,
    start_date: datetime.date,
    end_date: datetime.date,
    equipment_type_id: Optional[str],
) -> List[AvailableResource]:
    """
    Retrieve the Availability for all Sites in a Camp Area

    Sites are filtered on the provided date range and compatible
    equipment.

    Returns
    -------
    available_sites: List[AvailableResource]
        The list of available sites
    """
    search_filter = {
        "mapId": campground.map_id,
        "resourceLocationId": campground.facility_id,
        "bookingCategoryId": 0,
        "startDate": start_date.isoformat(),
        "endDate": end_date.isoformat(),
        "isReserving": True,
        "getDailyAvailability": False,
        "partySize": 1,
        "numEquipment": 1,
        "equipmentCategoryId": NON_GROUP_EQUIPMENT,
        "filterData": [],
    }
    if equipment_type_id:
        search_filter["subEquipmentCategoryId"] = equipment_type_id

    resources, additional_resources = self._find_matching_resources(
        campground.recreation_area_id, search_filter
    )

    # Resources are often deeply nested; fetch nested resources
    for map_id in additional_resources:
        search_filter["mapId"] = map_id
        avail, _ = self._find_matching_resources(
            campground.recreation_area_id, search_filter
        )
        resources.update(avail)

    availabilities = []
    for map_id, resource_details in resources.items():
        for resource_id, availability_details in resource_details.items():
            if availability_details[0]["availability"] == 0:
                ar = AvailableResource(resource_id=resource_id, map_id=map_id)
                availabilities.append(ar)

    return availabilities

rec_area_lookup(rec_area_id)

Lookup a recreation area by ID

Parameters:

Name	Type	Description	Default
rec_area_id	int	The recreation area ID to lookup	required

Returns:

Type	Description
domain_name, rec_ara: Tuple[str, RecreationArea]	The rec area's domain name and the recreation area object

Source code in camply/providers/going_to_camp/going_to_camp_provider.py

def rec_area_lookup(self, rec_area_id: int) -> Tuple[str, RecreationArea]:
    """
    Lookup a recreation area by ID

    Parameters
    ----------
    rec_area_id: int
        The recreation area ID to lookup

    Returns
    -------
    domain_name, rec_ara: Tuple[str, RecreationArea]
        The rec area's domain name and the recreation area object
    """
    for domain_name, rec_area in RECREATION_AREAS.items():
        if str(rec_area.recreation_area_id) == str(rec_area_id):
            return domain_name, rec_area

RecreationDotGov

Bases: RecreationDotGovBase

Recreation.gov: Campsite Searcher

Source code in camply/providers/recreation_dot_gov/recdotgov_camps.py

class RecreationDotGov(RecreationDotGovBase):
    """
    Recreation.gov: Campsite Searcher
    """

    facility_type = RIDBConfig.CAMPGROUND_FACILITY_FIELD_QUALIFIER
    resource_api_path = RIDBConfig.CAMPSITE_API_PATH
    activity_name = "CAMPING"
    api_response_class = CampsiteResponse
    api_base_path = RecreationBookingConfig.API_BASE_PATH
    api_search_result_class = RecDotGovCampsite
    api_search_result_key = "campsite_id"

    def paginate_recdotgov_campsites(
        self, facility_id: int, equipment: Optional[List[str]] = None
    ) -> List[RecDotGovCampsite]:
        """
        Paginate through the RecDotGov Campsite Metadata
        """
        results = 0
        continue_paginate = True
        endpoint_url = api_utils.generate_url(
            scheme=RecreationBookingConfig.API_SCHEME,
            netloc=RecreationBookingConfig.API_NET_LOC,
            path="api/search/campsites",
        )
        fq_list = [f"asset_id:{facility_id}"]
        if isinstance(equipment, list) and len(equipment) > 0:
            for item in equipment:
                fq_list.append(f"campsite_equipment_name:{item}")
        params = {
            "start": 0,
            "size": 1000,
            "fq": fq_list,
            "include_non_site_specific_campsites": True,
        }
        campsites: List[RecDotGovCampsite] = []
        while continue_paginate is True:
            response = self.make_recdotgov_request_retry(
                method="GET",
                url=endpoint_url,
                params=params,
            )
            returned_data = json.loads(response.content)
            campsite_response = RecDotGovCampsiteResponse(**returned_data)
            campsites += campsite_response.campsites
            results += campsite_response.size
            params.update(start=results)
            if results == campsite_response.total:
                continue_paginate = False
        return campsites

    def make_recdotgov_availability_request(
        self,
        campground_id: int,
        month: datetime,
    ) -> requests.Response:
        """
        Make a request to the RecreationDotGov API

        Parameters
        ----------
        campground_id
        month

        Returns
        -------
        requests.Response
        """
        api_endpoint = self._rec_availability_get_endpoint(
            path=f"{campground_id}/{RecreationBookingConfig.API_MONTH_PATH}"
        )
        formatted_month = month.strftime("%Y-%m-01T00:00:00.000Z")
        query_params = {"start_date": formatted_month}
        return self.make_recdotgov_request(
            method="GET",
            url=api_endpoint,
            params=query_params,
        )

    @classmethod
    def _items_to_unique_dicts(
        cls, item: Union[List[Dict[str, Any]], pd.Series]
    ) -> List[Dict[str, Any]]:
        """
        Ensure the proper items are parsed for equipment and attributes
        """
        if isinstance(item, pd.Series):
            list_of_dicts = list(chain.from_iterable(item.tolist()))
            unique_list_of_dicts = [
                dict(s) for s in {frozenset(d.items()) for d in list_of_dicts}
            ]
            return unique_list_of_dicts
        else:
            return item

    @classmethod
    def _get_equipment_attributes_location(
        cls,
        campsite_id: int,
        campsite_metadata: pd.DataFrame,
    ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], Optional[CampsiteLocation]]:
        """
        Index a DataFrame in a Complicated Way
        """
        try:
            equipment = campsite_metadata.at[campsite_id, "permitted_equipment"]
        except LookupError:
            equipment = None
        try:
            attributes = campsite_metadata.at[campsite_id, "attributes"]
        except LookupError:
            attributes = None
        try:
            latitude = campsite_metadata.at[campsite_id, "latitude"]
            longitude = campsite_metadata.at[campsite_id, "longitude"]
            location = CampsiteLocation(
                latitude=latitude,
                longitude=longitude,
            )
        except LookupError:
            location = None
        equipment = cls._items_to_unique_dicts(item=equipment)
        attributes = cls._items_to_unique_dicts(item=attributes)
        return equipment, attributes, location

    @classmethod
    def process_campsite_availability(
        cls,
        availability: Dict[str, Any],
        recreation_area: str,
        recreation_area_id: int,
        facility_name: str,
        facility_id: int,
        month: datetime,
        campsite_metadata: pd.DataFrame,
    ) -> List[Optional[AvailableCampsite]]:
        """
        Parse the JSON Response and return availabilities

        Parameters
        ----------
        availability: Dict[str, Any]
            API Response
        recreation_area: str
            Name of Recreation Area
        recreation_area_id: int
            ID of Recreation Area
        facility_name: str
            Campground Facility Name
        facility_id: int
            Campground Facility ID
        month: datetime
            Month to Process
        campsite_metadata: pd.DataFrame
            Metadata Fetched from the Recreation.gov API about the Campsites

        Returns
        -------
        total_campsite_availability: List[Optional[AvailableCampsite]]
            Any monthly availabilities
        """
        total_campsite_availability: List[Optional[AvailableCampsite]] = []
        campsite_data = CampsiteAvailabilityResponse(**availability)
        for campsite_id, site_related_data in campsite_data.campsites.items():
            for (
                matching_date,
                availability_status,
            ) in site_related_data.availabilities.items():
                if (
                    availability_status
                    not in RecreationBookingConfig.CAMPSITE_UNAVAILABLE_STRINGS
                ):
                    booking_url = (
                        f"{RecreationBookingConfig.CAMPSITE_BOOKING_URL}/{campsite_id}"
                    )
                    (
                        equipment,
                        attributes,
                        location,
                    ) = cls._get_equipment_attributes_location(
                        campsite_id=campsite_id, campsite_metadata=campsite_metadata
                    )
                    available_campsite = AvailableCampsite(
                        campsite_id=campsite_id,
                        booking_date=matching_date,
                        booking_end_date=matching_date + timedelta(days=1),
                        booking_nights=1,
                        campsite_site_name=site_related_data.site,
                        campsite_loop_name=site_related_data.loop,
                        campsite_type=site_related_data.campsite_type,
                        campsite_occupancy=(
                            site_related_data.min_num_people,
                            site_related_data.max_num_people,
                        ),
                        campsite_use_type=site_related_data.type_of_use,
                        availability_status=availability_status,
                        recreation_area=recreation_area,
                        recreation_area_id=recreation_area_id,
                        facility_name=facility_name,
                        facility_id=facility_id,
                        booking_url=booking_url,
                        permitted_equipment=equipment,
                        campsite_attributes=attributes,
                        location=location,
                    )
                    total_campsite_availability.append(available_campsite)
        return total_campsite_availability

make_recdotgov_availability_request(campground_id, month)

Make a request to the RecreationDotGov API

Parameters:

Name	Type	Description	Default
campground_id	int		required
month	datetime		required

Returns:

Type	Description
Response

Source code in camply/providers/recreation_dot_gov/recdotgov_camps.py

def make_recdotgov_availability_request(
    self,
    campground_id: int,
    month: datetime,
) -> requests.Response:
    """
    Make a request to the RecreationDotGov API

    Parameters
    ----------
    campground_id
    month

    Returns
    -------
    requests.Response
    """
    api_endpoint = self._rec_availability_get_endpoint(
        path=f"{campground_id}/{RecreationBookingConfig.API_MONTH_PATH}"
    )
    formatted_month = month.strftime("%Y-%m-01T00:00:00.000Z")
    query_params = {"start_date": formatted_month}
    return self.make_recdotgov_request(
        method="GET",
        url=api_endpoint,
        params=query_params,
    )

paginate_recdotgov_campsites(facility_id, equipment=None)

Paginate through the RecDotGov Campsite Metadata

Source code in camply/providers/recreation_dot_gov/recdotgov_camps.py

def paginate_recdotgov_campsites(
    self, facility_id: int, equipment: Optional[List[str]] = None
) -> List[RecDotGovCampsite]:
    """
    Paginate through the RecDotGov Campsite Metadata
    """
    results = 0
    continue_paginate = True
    endpoint_url = api_utils.generate_url(
        scheme=RecreationBookingConfig.API_SCHEME,
        netloc=RecreationBookingConfig.API_NET_LOC,
        path="api/search/campsites",
    )
    fq_list = [f"asset_id:{facility_id}"]
    if isinstance(equipment, list) and len(equipment) > 0:
        for item in equipment:
            fq_list.append(f"campsite_equipment_name:{item}")
    params = {
        "start": 0,
        "size": 1000,
        "fq": fq_list,
        "include_non_site_specific_campsites": True,
    }
    campsites: List[RecDotGovCampsite] = []
    while continue_paginate is True:
        response = self.make_recdotgov_request_retry(
            method="GET",
            url=endpoint_url,
            params=params,
        )
        returned_data = json.loads(response.content)
        campsite_response = RecDotGovCampsiteResponse(**returned_data)
        campsites += campsite_response.campsites
        results += campsite_response.size
        params.update(start=results)
        if results == campsite_response.total:
            continue_paginate = False
    return campsites

process_campsite_availability(availability, recreation_area, recreation_area_id, facility_name, facility_id, month, campsite_metadata) classmethod

Parse the JSON Response and return availabilities

Parameters:

Name	Type	Description	Default
availability	Dict[str, Any]	API Response	required
recreation_area	str	Name of Recreation Area	required
recreation_area_id	int	ID of Recreation Area	required
facility_name	str	Campground Facility Name	required
facility_id	int	Campground Facility ID	required
month	datetime	Month to Process	required
campsite_metadata	DataFrame	Metadata Fetched from the Recreation.gov API about the Campsites	required

Returns:

Name	Type	Description
total_campsite_availability	List[Optional[AvailableCampsite]]	Any monthly availabilities

Source code in camply/providers/recreation_dot_gov/recdotgov_camps.py

@classmethod
def process_campsite_availability(
    cls,
    availability: Dict[str, Any],
    recreation_area: str,
    recreation_area_id: int,
    facility_name: str,
    facility_id: int,
    month: datetime,
    campsite_metadata: pd.DataFrame,
) -> List[Optional[AvailableCampsite]]:
    """
    Parse the JSON Response and return availabilities

    Parameters
    ----------
    availability: Dict[str, Any]
        API Response
    recreation_area: str
        Name of Recreation Area
    recreation_area_id: int
        ID of Recreation Area
    facility_name: str
        Campground Facility Name
    facility_id: int
        Campground Facility ID
    month: datetime
        Month to Process
    campsite_metadata: pd.DataFrame
        Metadata Fetched from the Recreation.gov API about the Campsites

    Returns
    -------
    total_campsite_availability: List[Optional[AvailableCampsite]]
        Any monthly availabilities
    """
    total_campsite_availability: List[Optional[AvailableCampsite]] = []
    campsite_data = CampsiteAvailabilityResponse(**availability)
    for campsite_id, site_related_data in campsite_data.campsites.items():
        for (
            matching_date,
            availability_status,
        ) in site_related_data.availabilities.items():
            if (
                availability_status
                not in RecreationBookingConfig.CAMPSITE_UNAVAILABLE_STRINGS
            ):
                booking_url = (
                    f"{RecreationBookingConfig.CAMPSITE_BOOKING_URL}/{campsite_id}"
                )
                (
                    equipment,
                    attributes,
                    location,
                ) = cls._get_equipment_attributes_location(
                    campsite_id=campsite_id, campsite_metadata=campsite_metadata
                )
                available_campsite = AvailableCampsite(
                    campsite_id=campsite_id,
                    booking_date=matching_date,
                    booking_end_date=matching_date + timedelta(days=1),
                    booking_nights=1,
                    campsite_site_name=site_related_data.site,
                    campsite_loop_name=site_related_data.loop,
                    campsite_type=site_related_data.campsite_type,
                    campsite_occupancy=(
                        site_related_data.min_num_people,
                        site_related_data.max_num_people,
                    ),
                    campsite_use_type=site_related_data.type_of_use,
                    availability_status=availability_status,
                    recreation_area=recreation_area,
                    recreation_area_id=recreation_area_id,
                    facility_name=facility_name,
                    facility_id=facility_id,
                    booking_url=booking_url,
                    permitted_equipment=equipment,
                    campsite_attributes=attributes,
                    location=location,
                )
                total_campsite_availability.append(available_campsite)
    return total_campsite_availability

SearchRecreationDotGov

Bases: SearchRecreationDotGovBase

Searches on Recreation.gov for Campsites (default provider)

Source code in camply/search/search_recreationdotgov.py

class SearchRecreationDotGov(SearchRecreationDotGovBase):
    """
    Searches on Recreation.gov for Campsites (default provider)
    """

    provider_class = RecreationDotGov

SearchWindow

Bases: CamplyModel

Search Window for Campsite Search

Source code in camply/containers/data_containers.py

class SearchWindow(CamplyModel):
    """
    Search Window for Campsite Search
    """

    @validator("start_date")
    @classmethod
    def start_date_must_be_in_future(cls, v):
        """
        Validate that start_date is in the future.

        Coerece start date to today's date when it is not in the future.
        """
        current_date = datetime.datetime.now().date()
        if v < current_date:
            return current_date

        return v

    @validator("end_date")
    @classmethod
    def end_date_must_be_in_future(cls, v):
        """
        Validate that end_date is in the future
        """
        current_date = datetime.datetime.now().date()
        if v < current_date:
            raise ValueError("must be in the future")

        return v

    start_date: datetime.date
    end_date: datetime.date

    def get_date_range(self) -> List[datetime.date]:
        """
        Generate a List of Dates Between two Dates

        Returns
        -------
        List[datetime.date]
        """
        return [
            self.start_date + datetime.timedelta(days=x)
            for x in range((self.end_date - self.start_date).days)
        ]

    def get_current_start_date(self) -> datetime.date:
        """
        Return a start date with the current day in mind
        """
        return max((datetime.datetime.now().date(), self.start_date))

end_date_must_be_in_future(v) classmethod

Validate that end_date is in the future

Source code in camply/containers/data_containers.py

@validator("end_date")
@classmethod
def end_date_must_be_in_future(cls, v):
    """
    Validate that end_date is in the future
    """
    current_date = datetime.datetime.now().date()
    if v < current_date:
        raise ValueError("must be in the future")

    return v

get_current_start_date()

Return a start date with the current day in mind

Source code in camply/containers/data_containers.py

def get_current_start_date(self) -> datetime.date:
    """
    Return a start date with the current day in mind
    """
    return max((datetime.datetime.now().date(), self.start_date))

get_date_range()

Generate a List of Dates Between two Dates

Returns:

Type	Description
List[date]

Source code in camply/containers/data_containers.py

def get_date_range(self) -> List[datetime.date]:
    """
    Generate a List of Dates Between two Dates

    Returns
    -------
    List[datetime.date]
    """
    return [
        self.start_date + datetime.timedelta(days=x)
        for x in range((self.end_date - self.start_date).days)
    ]

start_date_must_be_in_future(v) classmethod

Validate that start_date is in the future.

Coerece start date to today's date when it is not in the future.

Source code in camply/containers/data_containers.py

@validator("start_date")
@classmethod
def start_date_must_be_in_future(cls, v):
    """
    Validate that start_date is in the future.

    Coerece start date to today's date when it is not in the future.
    """
    current_date = datetime.datetime.now().date()
    if v < current_date:
        return current_date

    return v

SearchYellowstone

Bases: BaseCampingSearch

Searches on YellowstoneNationalParkLodges.com for Campsites

Source code in camply/search/search_yellowstone.py

class SearchYellowstone(BaseCampingSearch):
    """
    Searches on YellowstoneNationalParkLodges.com for Campsites
    """

    recreation_area = Yellowstone.recreation_area
    provider_class = Yellowstone
    list_campsites_supported: bool = False

    # noinspection PyUnusedLocal
    def __init__(
        self,
        search_window: Union[SearchWindow, List[SearchWindow]],
        weekends_only: bool = False,
        campgrounds: Optional[Union[List[str], str]] = None,
        nights: int = 1,
        offline_search: bool = False,
        offline_search_path: Optional[str] = None,
        **kwargs,
    ) -> None:
        """
        Initialize with Search Parameters

        Parameters
        ----------
        search_window: Union[SearchWindow, List[SearchWindow]]
            Search Window tuple containing start date and End Date
        weekends_only: bool
            Whether to only search for Camping availabilities on the weekends (Friday /
            Saturday nights)
        campgrounds: Optional[Union[List[str], str]]
            Campground ID or List of Campground IDs
        nights: int
            minimum number of consecutive nights to search per campsite,defaults to 1
        offline_search: bool
            When set to True, the campsite search will both save the results of the
            campsites it's found, but also load those campsites before beginning a
            search for other campsites.
        offline_search_path: Optional[str]
            When offline search is set to True, this is the name of the file to be saved/loaded.
            When not specified, the filename will default to `camply_campsites.json`
        """
        super().__init__(
            search_window=search_window,
            weekends_only=weekends_only,
            nights=nights,
            offline_search=offline_search,
            offline_search_path=offline_search_path,
            **kwargs,
        )
        self.campgrounds = make_list(campgrounds)

    def get_all_campsites(self) -> List[AvailableCampsite]:
        """
        Search for all matching campsites in Yellowstone.

        Returns
        -------
        List[AvailableCampsite]
        """
        all_campsites = []
        searchable_campgrounds = self._get_searchable_campgrounds()
        this_month = datetime.now().date().replace(day=1)
        for month in self.search_months:
            if month >= this_month:
                all_campsites += self.campsite_finder.get_monthly_campsites(
                    month=month, nights=None if self.nights == 1 else self.nights
                )
        matching_campsites = self._filter_campsites_to_campgrounds(
            campsites=all_campsites, searchable_campgrounds=searchable_campgrounds
        )
        campsite_df = self.campsites_to_df(campsites=matching_campsites)
        campsite_df_validated = self._filter_date_overlap(campsites=campsite_df)
        time_window_end = max(self.search_days) + timedelta(days=1)
        compiled_campsite_df = campsite_df_validated[
            campsite_df_validated.booking_end_date <= pd.Timestamp(time_window_end)
        ]
        compiled_campsites = self.df_to_campsites(campsite_df=compiled_campsite_df)
        return compiled_campsites

    def _get_searchable_campgrounds(self) -> Optional[Set[str]]:
        """
        Return the Campgrounds for the Camping Search

        Returns
        -------
        Optional[Set[str]]
        """
        if self.campgrounds in [None, []]:
            return None
        supported_campsites = set(YellowstoneConfig.YELLOWSTONE_CAMPGROUNDS.keys())
        selected_campsites = set(self.campgrounds)
        searchable_campgrounds = supported_campsites.intersection(selected_campsites)
        if len(searchable_campgrounds) == 0:
            campground_ids = [
                f"`{key}` ({value})"
                for key, value in YellowstoneConfig.YELLOWSTONE_CAMPGROUNDS.items()
            ]
            error_message = (
                "You must supply a YellowstoneNationalParkLodges supported "
                "campground ID. Current supported Campground IDs: "
                f"{', '.join(campground_ids)}"
            )
            logger.error(error_message)
            raise SearchError(error_message)
        logger.info(f"{len(searchable_campgrounds)} Matching Campgrounds Found")
        for campground in searchable_campgrounds:
            logger.info(
                f"⛰  {YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_FORMAL_NAME} "
                f"(#{YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_ID}) - 🏕  "
                f"{YellowstoneConfig.YELLOWSTONE_CAMPGROUNDS[campground]} ({campground})"
            )
        return searchable_campgrounds

    def _filter_campsites_to_campgrounds(
        self, campsites: List[AvailableCampsite], searchable_campgrounds: Set[str]
    ) -> List[AvailableCampsite]:
        """
        Filter Campsites Down to Matching Campgrounds

        Parameters
        ----------
        campsites: List[AvailableCampsite]
        searchable_campgrounds: Set[str]

        Returns
        -------
        List[AvailableCampsite]
        """
        if self.campgrounds in [None, []]:
            return campsites
        matching_campsites = [
            campsite
            for campsite in campsites
            if campsite.facility_id in searchable_campgrounds
        ]
        return matching_campsites

    @classmethod
    def find_recreation_areas(cls, **kwargs) -> List[RecreationArea]:
        """
        Return the Yellowstone Recreation Area Object
        """
        log_sorted_response([cls.recreation_area])
        return [cls.recreation_area]

    def list_campsite_units(self) -> Any:
        """
        List Campsite Units

        Returns
        -------
        Any
        """
        raise NotImplementedError

__init__(search_window, weekends_only=False, campgrounds=None, nights=1, offline_search=False, offline_search_path=None, **kwargs)

Initialize with Search Parameters

Parameters:

Name	Type	Description	Default
search_window	Union[SearchWindow, List[SearchWindow]]	Search Window tuple containing start date and End Date	required
weekends_only	bool	Whether to only search for Camping availabilities on the weekends (Friday / Saturday nights)	False
campgrounds	Optional[Union[List[str], str]]	Campground ID or List of Campground IDs	None
nights	int	minimum number of consecutive nights to search per campsite,defaults to 1	1
offline_search	bool	When set to True, the campsite search will both save the results of the campsites it's found, but also load those campsites before beginning a search for other campsites.	False
offline_search_path	Optional[str]	When offline search is set to True, this is the name of the file to be saved/loaded. When not specified, the filename will default to camply_campsites.json	None

Source code in camply/search/search_yellowstone.py

def __init__(
    self,
    search_window: Union[SearchWindow, List[SearchWindow]],
    weekends_only: bool = False,
    campgrounds: Optional[Union[List[str], str]] = None,
    nights: int = 1,
    offline_search: bool = False,
    offline_search_path: Optional[str] = None,
    **kwargs,
) -> None:
    """
    Initialize with Search Parameters

    Parameters
    ----------
    search_window: Union[SearchWindow, List[SearchWindow]]
        Search Window tuple containing start date and End Date
    weekends_only: bool
        Whether to only search for Camping availabilities on the weekends (Friday /
        Saturday nights)
    campgrounds: Optional[Union[List[str], str]]
        Campground ID or List of Campground IDs
    nights: int
        minimum number of consecutive nights to search per campsite,defaults to 1
    offline_search: bool
        When set to True, the campsite search will both save the results of the
        campsites it's found, but also load those campsites before beginning a
        search for other campsites.
    offline_search_path: Optional[str]
        When offline search is set to True, this is the name of the file to be saved/loaded.
        When not specified, the filename will default to `camply_campsites.json`
    """
    super().__init__(
        search_window=search_window,
        weekends_only=weekends_only,
        nights=nights,
        offline_search=offline_search,
        offline_search_path=offline_search_path,
        **kwargs,
    )
    self.campgrounds = make_list(campgrounds)

find_recreation_areas(**kwargs) classmethod

Return the Yellowstone Recreation Area Object

Source code in camply/search/search_yellowstone.py

@classmethod
def find_recreation_areas(cls, **kwargs) -> List[RecreationArea]:
    """
    Return the Yellowstone Recreation Area Object
    """
    log_sorted_response([cls.recreation_area])
    return [cls.recreation_area]

get_all_campsites()

Search for all matching campsites in Yellowstone.

Returns:

Type	Description
List[AvailableCampsite]

Source code in camply/search/search_yellowstone.py

def get_all_campsites(self) -> List[AvailableCampsite]:
    """
    Search for all matching campsites in Yellowstone.

    Returns
    -------
    List[AvailableCampsite]
    """
    all_campsites = []
    searchable_campgrounds = self._get_searchable_campgrounds()
    this_month = datetime.now().date().replace(day=1)
    for month in self.search_months:
        if month >= this_month:
            all_campsites += self.campsite_finder.get_monthly_campsites(
                month=month, nights=None if self.nights == 1 else self.nights
            )
    matching_campsites = self._filter_campsites_to_campgrounds(
        campsites=all_campsites, searchable_campgrounds=searchable_campgrounds
    )
    campsite_df = self.campsites_to_df(campsites=matching_campsites)
    campsite_df_validated = self._filter_date_overlap(campsites=campsite_df)
    time_window_end = max(self.search_days) + timedelta(days=1)
    compiled_campsite_df = campsite_df_validated[
        campsite_df_validated.booking_end_date <= pd.Timestamp(time_window_end)
    ]
    compiled_campsites = self.df_to_campsites(campsite_df=compiled_campsite_df)
    return compiled_campsites

list_campsite_units()

List Campsite Units

Returns:

Type	Description
Any

Source code in camply/search/search_yellowstone.py

def list_campsite_units(self) -> Any:
    """
    List Campsite Units

    Returns
    -------
    Any
    """
    raise NotImplementedError

Yellowstone

Bases: BaseProvider

Scanner for Lodging in Yellowstone

Source code in camply/providers/xanterra/yellowstone_lodging.py

class Yellowstone(BaseProvider):
    """
    Scanner for Lodging in Yellowstone
    """

    recreation_area = RecreationArea(
        recreation_area=YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_FULL_NAME,
        recreation_area_id=YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_ID,
        recreation_area_location="USA",
    )

    def _get_monthly_availability(
        self, month: datetime, nights: Optional[int] = None
    ) -> dict:
        """
        Check All Lodging in Yellowstone for Campground Data

        Returns
        -------
        data_availability: dict
            Data Availability Dictionary
        """
        query_dict = {
            "date": self._ensure_current_month(month=month),
            "limit": 31,
            "rate_code": YellowstoneConfig.RATE_CODE,
        }
        if nights is not None:
            query_dict.update({"nights": nights})
        api_endpoint = self._get_api_endpoint(
            url_path=YellowstoneConfig.YELLOWSTONE_LODGING_PATH, query=None
        )
        logger.info(
            f"Searching for Yellowstone Lodging Availability: {month.strftime('%B, %Y')}"
        )
        all_resort_availability_data = self.make_yellowstone_request(
            endpoint=api_endpoint, params=query_dict
        )
        return all_resort_availability_data

    @staticmethod
    @tenacity.retry(
        wait=tenacity.wait_random_exponential(multiplier=3, max=1800),
        stop=tenacity.stop.stop_after_delay(6000),
    )
    def _try_retry_get_data(endpoint: str, params: Optional[dict] = None) -> dict:
        """
        Try and Retry Fetching Data from the Yellowstone API.

        Unfortunately this is a required method to request the data since the
        Yellowstone API doesn't always return data.

        Parameters
        ----------
        endpoint: str
            API Endpoint
        params

        Returns
        -------
        dict
        """
        yellowstone_headers = {}
        user_agent = {"User-Agent": UserAgent(browsers=["chrome"]).random}
        yellowstone_headers.update(user_agent)
        yellowstone_headers.update(STANDARD_HEADERS)
        yellowstone_headers.update(YellowstoneConfig.API_REFERRERS)
        response = requests.get(
            url=endpoint, headers=yellowstone_headers, params=params, timeout=30
        )
        if response.ok is True and response.text.strip() != "":
            return loads(response.content)
        else:
            error_message = (
                "Something went wrong with checking the "
                "Yellowstone Booking API. Will continue retrying."
            )
            logger.warning(error_message)
            raise RuntimeError(error_message)

    @staticmethod
    def make_yellowstone_request(endpoint: str, params: Optional[dict] = None) -> dict:
        """
        Try and Retry Fetching Data from the Yellowstone API.

        Unfortunately this is a required method to request the data since the
        Yellowstone API doesn't always return data.

        Parameters
        ----------
        endpoint: str
            API Endpoint
        params

        Returns
        -------
        dict
        """
        try:
            content = Yellowstone._try_retry_get_data(endpoint=endpoint, params=params)
        except RuntimeError as re:
            raise RuntimeError(f"error_message: {re}") from re
        return content

    @classmethod
    def _get_api_endpoint(cls, url_path: str, query: Optional[dict] = None) -> str:
        """
        Build the API Endpoint for All Yellowstone Lodging
        """
        if query is not None:
            query_string = parse.urlencode(query=query)
        else:
            query_string = ""
        url_components = {
            "scheme": YellowstoneConfig.API_SCHEME,
            "netloc": YellowstoneConfig.API_BASE_ENDPOINT,
            "url": url_path,
            "params": "",
            "query": query_string,
            "fragment": "",
        }
        api_endpoint = parse.urlunparse(tuple(url_components.values()))
        return api_endpoint

    @classmethod
    def _return_lodging_url(
        cls, lodging_code: str, month: datetime, params: Optional[dict] = ""
    ) -> str:
        """
        Return a Browser Loadable URL to book from

        Parameters
        ----------
        lodging_code: str
            Lodging Code from API
        month: datetime
            Month to return bookings filtered to
        params: Optional[dict]
            Optional URL Parameters

        Returns
        -------
        str
            URL String
        """
        query = {
            "dateFrom": month.strftime("%m-%d-%Y"),
            "adults": 1,
            "destination": lodging_code,
            "children": 0,
        }
        if params is not None:
            query.update(params)
        query_string = parse.urlencode(query=query)

        url_components = {
            "scheme": YellowstoneConfig.API_SCHEME,
            "netloc": YellowstoneConfig.WEBUI_BASE_ENDPOINT,
            "url": YellowstoneConfig.WEBUI_BOOKING_PATH,
            "params": "",
            "query": query_string,
            "fragment": "",
        }
        webui_endpoint = parse.urlunparse(tuple(url_components.values()))
        return webui_endpoint

    @classmethod
    def _compile_campground_availabilities(
        cls, availability: XantResortData
    ) -> List[dict]:
        """
        Gather Data about campground availabilities within a JSON Availability Objet

        Parameters
        ----------
        availability: ResortData
            JSON Availability Object

        Returns
        -------
        available_campsites:  List[dict]
            List of Availabilities as JSON
        """
        available_campsites = []
        for booking_date, daily_data in availability.availability.items():
            camping_keys = [
                key
                for key in daily_data.keys()
                if YellowstoneConfig.LODGING_CAMPGROUND_QUALIFIER in key
            ]
            for hotel_code in camping_keys:
                hotel_data = daily_data[hotel_code]
                try:
                    hotel_title = hotel_data.rates[YellowstoneConfig.RATE_CODE].title
                    hotel_rate_mins = hotel_data.rates[YellowstoneConfig.RATE_CODE].mins
                    if hotel_rate_mins != {1: 0}:
                        min_capacity = min(hotel_rate_mins.keys())
                        max_capacity = max(hotel_rate_mins.keys())
                        capacity = (min_capacity, max_capacity)
                        campsite = {
                            "campsite_id": None,
                            "booking_date": booking_date,
                            "campsite_occupancy": capacity,
                            "recreation_area": YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_NAME,
                            "recreation_area_id": YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_ID,
                            "facility_name": hotel_title.replace(
                                *YellowstoneConfig.YELLOWSTONE_CAMPGROUND_NAME_REPLACE
                            ),
                            "facility_id": hotel_code,
                        }
                        available_campsites.append(campsite)
                except KeyError:
                    pass
        logger.info(
            f"\t{logging_utils.get_emoji(available_campsites)}\t"
            f"{len(available_campsites)} sites found."
        )
        return available_campsites

    def _gather_campsite_specific_availability(
        self,
        available_campsites: List[dict],
        month: datetime,
        nights: Optional[int] = None,
    ) -> List[dict]:
        """
        Get campsite extra information

        Given a DataFrame of campsite availability, return updated Data with details
        about the actual campsites that are available (i.e Tent Size, RV Length, Etc)

        Parameters
        ----------
        available_campsites: List[dict]
            List of Available Campsites as JSON objects
        month: datetime
            Month object

        Returns
        -------
        List[dict]
        """
        available_room_array = []
        availability_df = DataFrame(data=available_campsites)
        if availability_df.empty is True:
            return available_room_array
        for facility_id, _facility_df in availability_df.groupby(
            YellowstoneConfig.FACILITY_ID
        ):
            api_endpoint = self._get_api_endpoint(
                url_path=YellowstoneConfig.YELLOWSTONE_CAMPSITE_AVAILABILITY, query=None
            )
            params = {"date": self._ensure_current_month(month=month), "limit": 31}
            if nights is not None:
                params.update({"nights": nights})
            campsite_data = self.make_yellowstone_request(
                endpoint=f"{api_endpoint}/{facility_id}", params=params
            )
            campsite_availability = campsite_data[
                YellowstoneConfig.BOOKING_AVAILABILITY
            ]
            booking_dates = campsite_availability.keys()
            availabilities = self._process_daily_availability(
                booking_dates=booking_dates,
                campsite_availability=campsite_availability,
                facility_id=facility_id,
            )
            available_room_array += availabilities
        return available_room_array

    @classmethod
    def _process_daily_availability(
        cls, booking_dates: List[str], campsite_availability: dict, facility_id: str
    ) -> List[dict]:
        """
        Process Monthly Availability

        Parameters
        ----------
        booking_dates: List[str]
            List of booking dates to process
        campsite_availability: dict
            Campsite availability dict
        facility_id: str
            Identification of the Facility

        Returns
        -------
        List[dict]
        """
        daily_availabilities = []
        for booking_date_str in booking_dates:
            daily_availability = campsite_availability[booking_date_str]
            if (
                daily_availability[YellowstoneConfig.FACILITY_STATUS]
                == YellowstoneConfig.FACILITY_STATUS_QUALIFIER
            ):
                available_rooms = daily_availability[YellowstoneConfig.FACILITY_ROOMS]
                for room in available_rooms:
                    if room[YellowstoneConfig.FACILITY_AVAILABLE_QUALIFIER] > 0:
                        daily_availabilities.append(
                            {
                                "booking_date": booking_date_str,
                                "facility_id": facility_id,
                                "campsite_code": room[
                                    YellowstoneConfig.FACILITY_ROOM_CODE
                                ],
                                "available": room[
                                    YellowstoneConfig.FACILITY_AVAILABLE_QUALIFIER
                                ],
                                "price": room[YellowstoneConfig.FACILITY_PRICE],
                            }
                        )
        return daily_availabilities

    def _get_property_information(self, available_rooms: List[dict]) -> List[dict]:
        """
        Gather Information About All Campgrounds / Hotels within Yellowstone

        Parameters
        ----------
        available_rooms: List[dict]

        Returns
        -------
        List[dict]
        """
        property_info_array = []
        availability_df = DataFrame(data=available_rooms)
        if availability_df.empty is True:
            return property_info_array
        facility_identifiers = availability_df[YellowstoneConfig.FACILITY_ID].unique()
        for facility_id in facility_identifiers:
            api_endpoint = self._get_api_endpoint(
                url_path=YellowstoneConfig.YELLOWSTONE_PROPERTY_INFO, query=None
            )
            campsite_info = self.make_yellowstone_request(
                endpoint=f"{api_endpoint}/{facility_id}"
            )
            campsite_codes = campsite_info.keys()
            for campsite_code in campsite_codes:
                campsite_data = campsite_info[campsite_code]
                property_info_array.append(
                    {
                        "facility_id": facility_id,
                        "campsite_code": campsite_code,
                        "campsite_title": campsite_data[
                            YellowstoneConfig.LODGING_TITLE
                        ],
                        "campsite_type": campsite_data[
                            YellowstoneConfig.FACILITY_TYPE
                        ].upper(),
                        "capacity": (
                            campsite_data[YellowstoneConfig.LODGING_OCCUPANCY_BASE],
                            campsite_data[YellowstoneConfig.LODGING_OCCUPANCY_MAX],
                        ),
                    }
                )
        return property_info_array

    def get_monthly_campsites(
        self, month: datetime, nights: Optional[int] = None
    ) -> List[AvailableCampsite]:
        """
        Return All Campsites Available in a Given Month

        Parameters
        ----------
        month: datetime
            Month to Search
        nights: Optional[int]
            Search for consecutive nights

        Returns
        -------
        List[AvailableCampsite]
        """
        now = datetime.now().date()
        search_date = month.replace(day=1)
        if month <= now:
            logger.info(
                "Cannot input search dates before today, adjusting search parameters."
            )
            search_date = search_date.replace(
                year=now.year, month=now.month, day=now.day
            )
        availability_found = self._get_monthly_availability(
            month=search_date, nights=nights
        )
        availability = XantResortData(**availability_found)
        monthly_campsites = self._compile_campground_availabilities(
            availability=availability
        )
        campsite_data = DataFrame(
            monthly_campsites, columns=YellowstoneConfig.CAMPSITE_DATA_COLUMNS
        ).drop_duplicates()
        if campsite_data.empty is True:
            return []
        available_room_array = self._gather_campsite_specific_availability(
            available_campsites=monthly_campsites, month=month, nights=nights
        )
        available_rooms = DataFrame(available_room_array)
        property_info = self._get_property_information(
            available_rooms=available_room_array
        )
        properties = DataFrame(property_info)
        merged_campsites = available_rooms.merge(
            properties,
            on=[
                YellowstoneConfig.FACILITY_ID_COLUMN,
                YellowstoneConfig.CAMPSITE_ID_COLUMN,
            ],
        )
        merged_campsites[YellowstoneConfig.BOOKING_DATE_COLUMN] = to_datetime(
            merged_campsites[YellowstoneConfig.BOOKING_DATE_COLUMN]
        )
        if nights is not None:
            nights_param = {"nights": nights}
        else:
            nights_param = {"nights": 1}
        booking_nights = nights_param.get("nights")
        merged_campsites[YellowstoneConfig.BOOKING_END_DATE_COLUMN] = merged_campsites[
            YellowstoneConfig.BOOKING_DATE_COLUMN
        ] + timedelta(days=booking_nights)
        merged_campsites[YellowstoneConfig.BOOKING_NIGHTS_COLUMN] = booking_nights
        final_campsites = merged_campsites.merge(
            campsite_data, on=YellowstoneConfig.FACILITY_ID_COLUMN
        ).sort_values(by=YellowstoneConfig.BOOKING_DATE_COLUMN)
        final_campsites[YellowstoneConfig.BOOKING_URL_COLUMN] = final_campsites.apply(
            lambda x: self._return_lodging_url(
                lodging_code=x.facility_id, month=x.booking_date, params=nights_param
            ),
            axis=1,
        )
        all_monthly_campsite_array = self._df_to_campsites(campsite_df=final_campsites)
        return all_monthly_campsite_array

    @classmethod
    def _df_to_campsites(cls, campsite_df: DataFrame) -> List[AvailableCampsite]:
        """
        Transform a DataFrame into an array of AvailableCampsites

        Parameters
        ----------
        campsite_df: DataFrame

        Returns
        -------
        List[AvailableCampsite]
        """
        all_monthly_campsite_array = []
        for _, row in campsite_df.iterrows():
            campsite = AvailableCampsite(
                campsite_id=row[YellowstoneConfig.CAMPSITE_ID_COLUMN],
                booking_date=row[YellowstoneConfig.BOOKING_DATE_COLUMN],
                booking_end_date=row[YellowstoneConfig.BOOKING_END_DATE_COLUMN],
                booking_nights=row[YellowstoneConfig.BOOKING_NIGHTS_COLUMN],
                campsite_site_name=row[YellowstoneConfig.CAMPSITE_SITE_NAME_COLUMN],
                campsite_loop_name=YellowstoneConfig.YELLOWSTONE_LOOP_NAME,
                campsite_type=row[YellowstoneConfig.CAMPSITE_TYPE_COLUMN],
                campsite_occupancy=row[YellowstoneConfig.CAMPSITE_OCCUPANCY_COLUMN],
                campsite_use_type=row[YellowstoneConfig.CAMPSITE_USE_TYPE_COLUMN],
                availability_status=YellowstoneConfig.CAMPSITE_AVAILABILITY_STATUS,
                recreation_area=YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_NAME,
                recreation_area_id=YellowstoneConfig.YELLOWSTONE_RECREATION_AREA_ID,
                facility_name=row[YellowstoneConfig.FACILITY_NAME_COLUMN],
                facility_id=row[YellowstoneConfig.FACILITY_ID_COLUMN],
                booking_url=row[YellowstoneConfig.BOOKING_URL_COLUMN],
            )
            all_monthly_campsite_array.append(campsite)
        return all_monthly_campsite_array

    @classmethod
    def _ensure_current_month(cls, month: datetime) -> datetime:
        """
        Ensure That We Never Give the Yellowstone API Dates in the past.

        Parameters
        ----------
        month: datetime

        Returns
        -------
        datetime
        """
        yellowstone_timezone = timezone(YellowstoneConfig.YELLOWSTONE_TIMEZONE)
        yellowstone_current_time = datetime.now(yellowstone_timezone).date()
        today = datetime(
            year=yellowstone_current_time.year,
            month=yellowstone_current_time.month,
            day=yellowstone_current_time.day,
        ).date()
        if today > month:
            month = today
        return month

    def find_campgrounds(self, **kwargs) -> List[CampgroundFacility]:
        """
        Print the Campgrounds inside of Yellowstone
        """
        log_sorted_response(YellowstoneConfig.YELLOWSTONE_CAMPGROUND_OBJECTS)
        return YellowstoneConfig.YELLOWSTONE_CAMPGROUND_OBJECTS

find_campgrounds(**kwargs)

Print the Campgrounds inside of Yellowstone

Source code in camply/providers/xanterra/yellowstone_lodging.py

def find_campgrounds(self, **kwargs) -> List[CampgroundFacility]:
    """
    Print the Campgrounds inside of Yellowstone
    """
    log_sorted_response(YellowstoneConfig.YELLOWSTONE_CAMPGROUND_OBJECTS)
    return YellowstoneConfig.YELLOWSTONE_CAMPGROUND_OBJECTS

get_monthly_campsites(month, nights=None)

Return All Campsites Available in a Given Month

Parameters:

Name	Type	Description	Default
month	datetime	Month to Search	required
nights	Optional[int]	Search for consecutive nights	None

Returns:

Type	Description
List[AvailableCampsite]

Source code in camply/providers/xanterra/yellowstone_lodging.py

def get_monthly_campsites(
    self, month: datetime, nights: Optional[int] = None
) -> List[AvailableCampsite]:
    """
    Return All Campsites Available in a Given Month

    Parameters
    ----------
    month: datetime
        Month to Search
    nights: Optional[int]
        Search for consecutive nights

    Returns
    -------
    List[AvailableCampsite]
    """
    now = datetime.now().date()
    search_date = month.replace(day=1)
    if month <= now:
        logger.info(
            "Cannot input search dates before today, adjusting search parameters."
        )
        search_date = search_date.replace(
            year=now.year, month=now.month, day=now.day
        )
    availability_found = self._get_monthly_availability(
        month=search_date, nights=nights
    )
    availability = XantResortData(**availability_found)
    monthly_campsites = self._compile_campground_availabilities(
        availability=availability
    )
    campsite_data = DataFrame(
        monthly_campsites, columns=YellowstoneConfig.CAMPSITE_DATA_COLUMNS
    ).drop_duplicates()
    if campsite_data.empty is True:
        return []
    available_room_array = self._gather_campsite_specific_availability(
        available_campsites=monthly_campsites, month=month, nights=nights
    )
    available_rooms = DataFrame(available_room_array)
    property_info = self._get_property_information(
        available_rooms=available_room_array
    )
    properties = DataFrame(property_info)
    merged_campsites = available_rooms.merge(
        properties,
        on=[
            YellowstoneConfig.FACILITY_ID_COLUMN,
            YellowstoneConfig.CAMPSITE_ID_COLUMN,
        ],
    )
    merged_campsites[YellowstoneConfig.BOOKING_DATE_COLUMN] = to_datetime(
        merged_campsites[YellowstoneConfig.BOOKING_DATE_COLUMN]
    )
    if nights is not None:
        nights_param = {"nights": nights}
    else:
        nights_param = {"nights": 1}
    booking_nights = nights_param.get("nights")
    merged_campsites[YellowstoneConfig.BOOKING_END_DATE_COLUMN] = merged_campsites[
        YellowstoneConfig.BOOKING_DATE_COLUMN
    ] + timedelta(days=booking_nights)
    merged_campsites[YellowstoneConfig.BOOKING_NIGHTS_COLUMN] = booking_nights
    final_campsites = merged_campsites.merge(
        campsite_data, on=YellowstoneConfig.FACILITY_ID_COLUMN
    ).sort_values(by=YellowstoneConfig.BOOKING_DATE_COLUMN)
    final_campsites[YellowstoneConfig.BOOKING_URL_COLUMN] = final_campsites.apply(
        lambda x: self._return_lodging_url(
            lodging_code=x.facility_id, month=x.booking_date, params=nights_param
        ),
        axis=1,
    )
    all_monthly_campsite_array = self._df_to_campsites(campsite_df=final_campsites)
    return all_monthly_campsite_array

make_yellowstone_request(endpoint, params=None) staticmethod

Try and Retry Fetching Data from the Yellowstone API.

Unfortunately this is a required method to request the data since the Yellowstone API doesn't always return data.

Parameters:

Name	Type	Description	Default
endpoint	str	API Endpoint	required
params	Optional[dict]		None

Returns:

Type	Description
dict

Source code in camply/providers/xanterra/yellowstone_lodging.py

@staticmethod
def make_yellowstone_request(endpoint: str, params: Optional[dict] = None) -> dict:
    """
    Try and Retry Fetching Data from the Yellowstone API.

    Unfortunately this is a required method to request the data since the
    Yellowstone API doesn't always return data.

    Parameters
    ----------
    endpoint: str
        API Endpoint
    params

    Returns
    -------
    dict
    """
    try:
        content = Yellowstone._try_retry_get_data(endpoint=endpoint, params=params)
    except RuntimeError as re:
        raise RuntimeError(f"error_message: {re}") from re
    return content