Skip to content

recdotgov_camps

Campsite Searching: Recreation.gov

RecreationDotGov #

Bases: RecreationDotGovBase

Recreation.gov: Campsite Searcher

Source code in camply/providers/recreation_dot_gov/recdotgov_camps.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
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 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
@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