Skip to content

providers

providers init file

AlabamaStateParks #

Bases: UseDirectProvider

Alabama State Parks

Source code in camply/providers/usedirect/variations.py 
10
11
12
13
14
15
16
17
18
19
20
class AlabamaStateParks(UseDirectProvider):
    """
    Alabama State Parks
    """

    base_url = "https://alabamardr.usedirect.com"
    campground_url = "https://www.reservealapark.com"
    rdr_path = "alabamardr"
    booking_path = "AlabamaWebHome/Facilities/SearchViewUnitAvailabity.aspx"
    booking_path_params = False
    state_code = "AL"

ArizonaStateParks #

Bases: UseDirectProvider

Arizona State Parks

Source code in camply/providers/usedirect/variations.py 
69
70
71
72
73
74
75
76
77
78
class ArizonaStateParks(UseDirectProvider):
    """
    Arizona State Parks
    """

    base_url = "https://azrdr.usedirect.com"
    campground_url = "https://ArizonaStateParks.com"
    rdr_path = "azrdr"
    booking_path = "reserve"
    state_code = "AZ"

BaseProvider #

Bases: ABC

Base Provider Class

Source code in camply/providers/base_provider.py 
 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
class BaseProvider(ABC):
    """
    Base Provider Class
    """

    RETRY_CONFIG: Type[APIConfig] = APIConfig
    FIVE_HUNDRED_STATUS_CODES = [
        # Official Server Errors
        500,  # Internal Server Error
        501,  # Not Implemented
        502,  # Bad Gateway
        503,  # Service Unavailable
        504,  # Gateway Timeout
        505,  # HTTP Version Not Supported
        506,  # Variant Also Negotiates
        507,  # Insufficient Storage
        508,  # Loop Detected
        510,  # Not Extended
        511,  # Network Authentication Required
        # Unofficial Server Errors
        509,  # Bandwidth Limit Exceeded
        529,  # Site is overloaded
        530,  # Site is frozen
        598,  # Network read timeout error
        599,  # Network connect timeout error
        # Vendor Errors
        520,  # Unknown Error
        521,  # Web Server Is Down
        522,  # Connection Timed Out
        523,  # Origin Is Unreachable
        524,  # A Timeout Occurred
        525,  # SSL Handshake Failed
        526,  # Invalid SSL Certificate
        527,  # Railgun Error
        530,  # Origin DNS Error
        561,  # Unauthorized
    ]

    def __repr__(self):
        """
        String Representation

        Returns
        -------
        str
        """
        return f"<{self.__class__.__name__}>"

    def __init__(self):
        """
        Initialize with a session
        """
        _user_agent = UserAgent(browsers=["chrome"]).random
        self.session = requests.Session()
        self.headers = {"User-Agent": _user_agent}
        self.session.headers = self.headers
        self.json_headers = self.headers.copy()
        self.json_headers.update({"Content-Type": "application/json"})

    @classmethod
    def get_search_months(cls, search_days) -> List[datetime]:
        """
        Get the Unique Months that need to be Searched

        Returns
        -------
        search_months: List[datetime]
            Datetime Months to search for reservations
        """
        truncated_months = {day.replace(day=1) for day in search_days}
        if len(truncated_months) > 1:
            logger.info(
                f"{len(truncated_months)} different months selected for search, "
                f"ranging from {min(search_days)} to {max(search_days)}"
            )
            return sorted(truncated_months)
        elif len(truncated_months) == 0:
            logger.error(SearchConfig.ERROR_MESSAGE)
            raise RuntimeError(SearchConfig.ERROR_MESSAGE)
        else:
            return sorted(truncated_months)

    @abstractmethod
    def find_campgrounds(self) -> List[CampgroundFacility]:
        """
        List Recreation Areas for the provider
        """

    def make_http_request(
        self,
        url: str,
        method: str = "GET",
        data: Optional[Union[Dict[str, Any], str]] = None,
        headers: Optional[Dict[str, Any]] = None,
        retry_response_codes: Optional[List[int]] = None,
    ) -> requests.Response:
        """
        Make an HTTP Request

        Parameters
        ----------
        url: str
            URL to make the request to
        method: str
            HTTP Method to use. Defaults to GET
        data: Optional[Union[Dict[str, Any], str]]
            Data to send with the request
        headers: Optional[Dict[str, Any]]
            Headers to send with the request
        retry_response_codes: Optional[List[int]]
            List of response codes to raise a ProviderError. on. Defaults to 500 range

        Returns
        -------
        response: requests.Response

        Raises
        ------
        ProviderError
            If the response code is in the retry_response_codes list
        HTTPError
            If the response code is not in the retry_response_codes list and the request fails
        """
        if retry_response_codes is None:
            retry_response_codes = self.FIVE_HUNDRED_STATUS_CODES
        response = self.session.request(
            method=method, url=url, data=data, headers=headers
        )
        if response.status_code not in retry_response_codes:
            response.raise_for_status()
        else:
            error_message = f"HTTP Error - {self.__class__.__name__} - {response.url} - {response.status_code}"
            logger.warning(error_message)
            error_message += f": {response.text}"
            raise ProviderError(error_message)
        return response

    def make_http_request_retry(
        self,
        url: str,
        method: str = "GET",
        data: Optional[Union[Dict[str, Any], str]] = None,
        headers: Optional[Dict[str, Any]] = None,
        retry_response_codes: Optional[List[int]] = None,
    ) -> requests.Response:
        """
        Make an HTTP Request with Exponential Backoff

        This method will retry the request with exponential backoff if the request fails.
        By default, it will only ignore 500 range status codes, but this can be overridden.

        Parameters
        ----------
        url: str
            URL to make the request to
        method: str
            HTTP Method to use. Defaults to GET
        data: Optional[Union[Dict[str, Any], str]]
            Data to send with the request
        headers: Optional[Dict[str, Any]]
            Headers to send with the request
        retry_response_codes: Optional[List[int]]
            List of response codes to retry on. Defaults to 500 range

        Returns
        -------
        response: requests.Response
        """
        retryer = tenacity.Retrying(
            wait=tenacity.wait_random_exponential(
                multiplier=self.RETRY_CONFIG.RETRY_API_MULTIPLIER,
                max=self.RETRY_CONFIG.RETRY_MAX_API_ATTEMPTS,
            ),
            stop=tenacity.stop.stop_after_delay(
                self.RETRY_CONFIG.RETRY_MAX_API_TIMEOUT
            ),
            retry=tenacity.retry_if_exception_type(ProviderError),
        )
        response: requests.Response = retryer.__call__(
            fn=self.make_http_request,
            url=url,
            method=method,
            data=data,
            headers=headers,
            retry_response_codes=retry_response_codes,
        )
        return response

__init__() #

Initialize with a session

Source code in camply/providers/base_provider.py 
81
82
83
84
85
86
87
88
89
90
def __init__(self):
    """
    Initialize with a session
    """
    _user_agent = UserAgent(browsers=["chrome"]).random
    self.session = requests.Session()
    self.headers = {"User-Agent": _user_agent}
    self.session.headers = self.headers
    self.json_headers = self.headers.copy()
    self.json_headers.update({"Content-Type": "application/json"})

__repr__() #

String Representation

Returns:

Type Description
str
Source code in camply/providers/base_provider.py 
71
72
73
74
75
76
77
78
79
def __repr__(self):
    """
    String Representation

    Returns
    -------
    str
    """
    return f"<{self.__class__.__name__}>"

find_campgrounds() abstractmethod #

List Recreation Areas for the provider

Source code in camply/providers/base_provider.py 
115
116
117
118
119
@abstractmethod
def find_campgrounds(self) -> List[CampgroundFacility]:
    """
    List Recreation Areas for the provider
    """

get_search_months(search_days) classmethod #

Get the Unique Months that need to be Searched

Returns:

Name Type Description
search_months List[datetime]

Datetime Months to search for reservations
Source code in camply/providers/base_provider.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@classmethod
def get_search_months(cls, search_days) -> List[datetime]:
    """
    Get the Unique Months that need to be Searched

    Returns
    -------
    search_months: List[datetime]
        Datetime Months to search for reservations
    """
    truncated_months = {day.replace(day=1) for day in search_days}
    if len(truncated_months) > 1:
        logger.info(
            f"{len(truncated_months)} different months selected for search, "
            f"ranging from {min(search_days)} to {max(search_days)}"
        )
        return sorted(truncated_months)
    elif len(truncated_months) == 0:
        logger.error(SearchConfig.ERROR_MESSAGE)
        raise RuntimeError(SearchConfig.ERROR_MESSAGE)
    else:
        return sorted(truncated_months)

make_http_request(url, method='GET', data=None, headers=None, retry_response_codes=None) #

Make an HTTP Request

Parameters:

Name Type Description Default
url str

URL to make the request to

 required
method str

HTTP Method to use. Defaults to GET

 'GET'
data Optional[Union[Dict[str, Any], str]]

Data to send with the request

 None
headers Optional[Dict[str, Any]]

Headers to send with the request

 None
retry_response_codes Optional[List[int]]

List of response codes to raise a ProviderError. on. Defaults to 500 range

 None

Returns:

Name Type Description
response Response

Raises:

Type Description
ProviderError

If the response code is in the retry_response_codes list
HTTPError

If the response code is not in the retry_response_codes list and the request fails
Source code in camply/providers/base_provider.py 
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
def make_http_request(
    self,
    url: str,
    method: str = "GET",
    data: Optional[Union[Dict[str, Any], str]] = None,
    headers: Optional[Dict[str, Any]] = None,
    retry_response_codes: Optional[List[int]] = None,
) -> requests.Response:
    """
    Make an HTTP Request

    Parameters
    ----------
    url: str
        URL to make the request to
    method: str
        HTTP Method to use. Defaults to GET
    data: Optional[Union[Dict[str, Any], str]]
        Data to send with the request
    headers: Optional[Dict[str, Any]]
        Headers to send with the request
    retry_response_codes: Optional[List[int]]
        List of response codes to raise a ProviderError. on. Defaults to 500 range

    Returns
    -------
    response: requests.Response

    Raises
    ------
    ProviderError
        If the response code is in the retry_response_codes list
    HTTPError
        If the response code is not in the retry_response_codes list and the request fails
    """
    if retry_response_codes is None:
        retry_response_codes = self.FIVE_HUNDRED_STATUS_CODES
    response = self.session.request(
        method=method, url=url, data=data, headers=headers
    )
    if response.status_code not in retry_response_codes:
        response.raise_for_status()
    else:
        error_message = f"HTTP Error - {self.__class__.__name__} - {response.url} - {response.status_code}"
        logger.warning(error_message)
        error_message += f": {response.text}"
        raise ProviderError(error_message)
    return response

make_http_request_retry(url, method='GET', data=None, headers=None, retry_response_codes=None) #

Make an HTTP Request with Exponential Backoff

This method will retry the request with exponential backoff if the request fails. By default, it will only ignore 500 range status codes, but this can be overridden.

Parameters:

Name Type Description Default
url str

URL to make the request to

 required
method str

HTTP Method to use. Defaults to GET

 'GET'
data Optional[Union[Dict[str, Any], str]]

Data to send with the request

 None
headers Optional[Dict[str, Any]]

Headers to send with the request

 None
retry_response_codes Optional[List[int]]

List of response codes to retry on. Defaults to 500 range

 None

Returns:

Name Type Description
response Response
Source code in camply/providers/base_provider.py 
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
def make_http_request_retry(
    self,
    url: str,
    method: str = "GET",
    data: Optional[Union[Dict[str, Any], str]] = None,
    headers: Optional[Dict[str, Any]] = None,
    retry_response_codes: Optional[List[int]] = None,
) -> requests.Response:
    """
    Make an HTTP Request with Exponential Backoff

    This method will retry the request with exponential backoff if the request fails.
    By default, it will only ignore 500 range status codes, but this can be overridden.

    Parameters
    ----------
    url: str
        URL to make the request to
    method: str
        HTTP Method to use. Defaults to GET
    data: Optional[Union[Dict[str, Any], str]]
        Data to send with the request
    headers: Optional[Dict[str, Any]]
        Headers to send with the request
    retry_response_codes: Optional[List[int]]
        List of response codes to retry on. Defaults to 500 range

    Returns
    -------
    response: requests.Response
    """
    retryer = tenacity.Retrying(
        wait=tenacity.wait_random_exponential(
            multiplier=self.RETRY_CONFIG.RETRY_API_MULTIPLIER,
            max=self.RETRY_CONFIG.RETRY_MAX_API_ATTEMPTS,
        ),
        stop=tenacity.stop.stop_after_delay(
            self.RETRY_CONFIG.RETRY_MAX_API_TIMEOUT
        ),
        retry=tenacity.retry_if_exception_type(ProviderError),
    )
    response: requests.Response = retryer.__call__(
        fn=self.make_http_request,
        url=url,
        method=method,
        data=data,
        headers=headers,
        retry_response_codes=retry_response_codes,
    )
    return response

FairfaxCountyParks #

Bases: UseDirectProvider

Fairfax County Parks

Source code in camply/providers/usedirect/variations.py 
121
122
123
124
125
126
127
128
129
130
131
class FairfaxCountyParks(UseDirectProvider):
    """
    Fairfax County Parks
    """

    base_url = "https://fairfaxrdr.usedirect.com"
    campground_url = "https://fairfax.usedirect.com"
    rdr_path = "FCPARDR"
    booking_path = "FairfaxFCPAWeb/Facilities/SearchViewUnitAvailabity.aspx"
    booking_path_params = False
    state_code = "VA"

FloridaStateParks #

Bases: UseDirectProvider

Florida State Parks

Source code in camply/providers/usedirect/variations.py 
33
34
35
36
37
38
39
40
41
42
class FloridaStateParks(UseDirectProvider):
    """
    Florida State Parks
    """

    base_url = "https://floridardr.usedirect.com"
    campground_url = "https://www.reserve.floridastateparks.org"
    rdr_path = "FloridaRDR"
    booking_path = "Web"
    state_code = "FL"

GoingToCamp #

Bases: BaseProvider

Going To Camp API provider

Source code in camply/providers/going_to_camp/going_to_camp_provider.py 
 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
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
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)
        user_agent = {"User-Agent": UserAgent(browsers=["chrome"]).random}
        response = requests.get(url=url, headers=user_agent, 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 
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
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 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
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 
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
@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

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 
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
237
238
239
240
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 
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
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 
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
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 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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

MaricopaCountyParks #

Bases: UseDirectProvider

Maricopa County Parks

Source code in camply/providers/usedirect/variations.py 
108
109
110
111
112
113
114
115
116
117
118
class MaricopaCountyParks(UseDirectProvider):
    """
    Maricopa County Parks
    """

    base_url = "https://maricopardr.usedirect.com"
    campground_url = "https://www.maricopacountyparks.org"
    rdr_path = "maricopardr"
    booking_path = "MaricopaWeb/Facilities/SearchViewUnitAvailabity.aspx"
    booking_path_params = False
    state_code = "AZ"

MinnesotaStateParks #

Bases: UseDirectProvider

Minnesota State Parks

Source code in camply/providers/usedirect/variations.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
class MinnesotaStateParks(UseDirectProvider):
    """
    Minnesota State Parks
    """

    base_url = "https://mnrdr.usedirect.com"
    campground_url = "https://reservemn.usedirect.com"
    rdr_path = "minnesotardr"
    booking_path = "MinnesotaWeb"
    state_code = "MN"

MissouriStateParks #

Bases: UseDirectProvider

Missouri State Parks

Source code in camply/providers/usedirect/variations.py 
81
82
83
84
85
86
87
88
89
90
class MissouriStateParks(UseDirectProvider):
    """
    Missouri State Parks
    """

    base_url = "https://msprdr.usedirect.com"
    campground_url = "https://icampmo1.usedirect.com"
    rdr_path = "msprdr"
    booking_path = "MSPWeb"
    state_code = "MO"

NorthernTerritory #

Bases: UseDirectProvider

Australian NorthernTerritory

Source code in camply/providers/usedirect/variations.py 
150
151
152
153
154
155
156
157
158
159
160
class NorthernTerritory(UseDirectProvider):
    """
    Australian NorthernTerritory
    """

    base_url = "https://northernterritoryrdr.usedirect.com"
    campground_url = "https://parkbookings.nt.gov.au"
    rdr_path = "NorthernTerritoryRDR"
    booking_path = "Web/Facilities/SearchViewUnitAvailabity.aspx"
    booking_path_params = False
    state_code = "NT"

OhioStateParks #

Bases: UseDirectProvider

Ohio State Parks

Source code in camply/providers/usedirect/variations.py 
45
46
47
48
49
50
51
52
53
54
class OhioStateParks(UseDirectProvider):
    """
    Ohio State Parks
    """

    base_url = "https://ohiordr.usedirect.com"
    campground_url = "https://www.OhioStateParks.com"
    rdr_path = "ohiordr"
    booking_path = "OhioCampWeb"
    state_code = "OH"

OregonMetro #

Bases: UseDirectProvider

Oregon Metro Parks

Source code in camply/providers/usedirect/variations.py 
134
135
136
137
138
139
140
141
142
143
144
class OregonMetro(UseDirectProvider):
    """
    Oregon Metro Parks
    """

    base_url = "https://oregonrdr.usedirect.com"
    campground_url = "https://reservemetro.usedirect.com"
    rdr_path = "oregonmetrordr"
    booking_path = "MetroWeb/Facilities/SearchViewUnitAvailabity.aspx"
    booking_path_params = False
    state_code = "OR"

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

RecreationDotGovDailyTicket #

Bases: RecreationDotGovDailyMixin, RecreationDotGovTicket

RecreationDotGovTicket: Daily

Daily MixIn for Tickets

Source code in camply/providers/recreation_dot_gov/recdotgov_tours.py 
468
469
470
471
472
473
class RecreationDotGovDailyTicket(RecreationDotGovDailyMixin, RecreationDotGovTicket):
    """
    RecreationDotGovTicket: Daily

    Daily MixIn for Tickets
    """

RecreationDotGovDailyTimedEntry #

Bases: RecreationDotGovDailyMixin, RecreationDotGovTimedEntry

RecreationDotGovTimedEntry: Daily

Daily MixIn for Tours

Source code in camply/providers/recreation_dot_gov/recdotgov_tours.py 
476
477
478
479
480
481
482
483
class RecreationDotGovDailyTimedEntry(
    RecreationDotGovDailyMixin, RecreationDotGovTimedEntry
):
    """
    RecreationDotGovTimedEntry: Daily

    Daily MixIn for Tours
    """

RecreationDotGovTicket #

Bases: RecreationDotGovTours

RecreationDotGovTicket

Tickets for Tours

Source code in camply/providers/recreation_dot_gov/recdotgov_tours.py 
440
441
442
443
444
445
446
447
448
449
450
class RecreationDotGovTicket(RecreationDotGovTours):
    """
    RecreationDotGovTicket

    Tickets for Tours
    """

    facility_type = RIDBConfig.TICKET_FACILITY_FIELD_QUALIFIER
    api_search_fq = "entity_type:tour"
    api_base_path = "api/ticket/availability/facility/"
    booking_url = "https://www.recreation.gov/ticket/{facility_id}/ticket/{tour_id}"

RecreationDotGovTimedEntry #

Bases: RecreationDotGovTours

RecreationDotGovTimedEntry

Timed Entries

Source code in camply/providers/recreation_dot_gov/recdotgov_tours.py 
453
454
455
456
457
458
459
460
461
462
463
464
465
class RecreationDotGovTimedEntry(RecreationDotGovTours):
    """
    RecreationDotGovTimedEntry

    Timed Entries
    """

    facility_type = RIDBConfig.TIMED_ENTRY_FACILITY_FIELD_QUALIFIER
    api_search_fq = "entity_type:timedentry_tour"
    api_base_path = "api/timedentry/availability/facility/"
    booking_url = (
        "https://www.recreation.gov/timed-entry/{facility_id}/ticket/{tour_id}"
    )

ReserveCalifornia #

Bases: UseDirectProvider

ReserveCalifornia

Source code in camply/providers/usedirect/variations.py 
23
24
25
26
27
28
29
30
class ReserveCalifornia(UseDirectProvider):
    """
    ReserveCalifornia
    """

    base_url = "https://calirdr.usedirect.com"
    campground_url = "https://www.reservecalifornia.com"
    state_code = "CA"

VirginiaStateParks #

Bases: UseDirectProvider

Virginia State Parks

Source code in camply/providers/usedirect/variations.py 
57
58
59
60
61
62
63
64
65
66
class VirginiaStateParks(UseDirectProvider):
    """
    Virginia State Parks
    """

    base_url = "https://virginiardr.usedirect.com"
    campground_url = "https://VirginiaStateParks.com"
    rdr_path = "virginiardr"
    booking_path = "Web"
    state_code = "VA"

Yellowstone #

Bases: BaseProvider

Scanner for Lodging in Yellowstone

Source code in camply/providers/xanterra/yellowstone_lodging.py 
 28
 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
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
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 
524
525
526
527
528
529
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 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
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 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@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