nautical.io package

Submodules

nautical.io.buoy module

nautical.io.buoy.create_buoy(buoy: str) → Buoy | None[source]

Provide a full workup for a specific buoy. If the buoy is None or it cannot be found then the data returned will be considered invalid as None

Parameters:

buoy – id of the buoy to do a workup on

Returns:

BuoyWorkup if successful else None

nautical.io.buoy.fill_buoy(buoy: Buoy) → None[source]

Pass in a Buoy object that needs to be filled in with the current data. The buoy object will have the validity set if the results were successful

Parameters:

buoy – nautical.noaa.buoy.Buoy object

nautical.io.buoy.get_current_data(soup: BeautifulSoup, buoy: BuoyData, search: str | list[str]) → bool[source]

Search the beautiful soup object for a TABLE containing the search string. The function will grab the data from the table and create a NOAAData object and return the data

Parameters:

  • soup – beautiful soup object generated from the get_url_source()

  • buoy – BuoyData object that should be filled with data as this function parses the data.

  • search – text to search for in the soup object.

Returns:

True when data has been found and set

nautical.io.buoy.get_buoy_data(soup: BeautifulSoup, buoy: BuoyData, search: str | list[str]) → bool

Search the beautiful soup object for a TABLE containing the search string. The function will grab the data from the table and create a NOAAData object and return the data

Parameters:

  • soup – beautiful soup object generated from the get_url_source()

  • buoy – BuoyData object that should be filled with data as this function parses the data.

  • search – text to search for in the soup object.

Returns:

True when data has been found and set

nautical.io.cdata module

nautical.io.cdata.parse_winds(wind_data: str) → dict[str, float | None][source]

Parse the wind information. The wind data includes a direction, speed in knots, as well as the gust speed

Parameters:

wind_data – String containing the wind string

Returns:

Dictionary containing the windspeed and gust information

nautical.io.cdata.parse_location(location_data: str) → dict[source]

Parse the latitude and longitude values out of the the string that was passed in. The latitude and longitude should contain the NSEW strings describing their sign.

Parameters:

location_data – String data contain latitude and longitude values

Returns:

dictionary containing the location point (when valid)

nautical.io.cdata.parse_time(time_data: str) → dict[source]

Parse the month/day/year time out of the time data string from CDATA.

Parameters:

time_data – String containing all time and date information

Returns:

dictionary mm, dd, year, time where time is NauticalTime

nautical.io.cdata.parse_cdata(cdata: str) → dict[source]

Parse the CDATA string that contains information about this presumed buoy. The data here will be used as supplemental information as some buoys cannot be scraped online (no valid or useable data).

Parameters:

cdata – String containing the CDATA or description element from kml

Returns:

Dictionary containing all parsed fields.

nautical.io.cdata.fill_buoy_with_cdata(buoy: Buoy, cdata: str) → None[source]

Parse the CDATA string that contains information about this presumed buoy. The data here will be used as supplemental information as some buoys cannot be scraped online (no valid or useable data).

Parameters:

  • buoy – nautical.noaa.buoy.Buoy

  • cdata – String containing the CDATA or description element from kml

Returns:

Buoy object

nautical.io.retry module

Retry logic and rate limiting for web requests.

This module provides intelligent retry mechanisms with exponential backoff, rate limit detection, and respect for HTTP Retry-After headers.

class nautical.io.retry.RetryConfig(max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 60.0, backoff_factor: float = 2.0, retry_on_status: set[int] | None = None, respect_retry_after: bool = True)[source]

Bases: object

Configuration for retry behavior.

max_retries

Maximum number of retry attempts (default: 3)

initial_delay

Initial delay in seconds before first retry (default: 1.0)

max_delay

Maximum delay between retries in seconds (default: 60.0)

backoff_factor

Multiplier for exponential backoff (default: 2.0)

retry_on_status

HTTP status codes that should trigger retry (default: 429, 500-599)

respect_retry_after

Whether to respect HTTP Retry-After header (default: True)

__init__(max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 60.0, backoff_factor: float = 2.0, retry_on_status: set[int] | None = None, respect_retry_after: bool = True) → None[source]
class nautical.io.retry.RateLimiter(requests_per_window: int = 30, window_seconds: int = 60)[source]

Bases: object

Simple rate limiter to prevent overwhelming the server.

Uses a token bucket algorithm to limit requests per time window.

requests_per_window

Number of requests allowed per time window

window_seconds

Time window in seconds (default: 60)

__init__(requests_per_window: int = 30, window_seconds: int = 60) → None[source]

Initialize rate limiter.

Parameters:

  • requests_per_window – Max requests allowed in the time window

  • window_seconds – Length of the time window in seconds

acquire() → None[source]

Acquire a token, blocking if necessary.

This method will sleep if no tokens are available, waiting until a token becomes available.

nautical.io.retry.get_retry_delay(attempt: int, config: RetryConfig, retry_after: str | int | None = None) → float[source]

Calculate delay before next retry attempt.

Parameters:

  • attempt – Current attempt number (0-indexed)

  • config – RetryConfig instance

  • retry_after – Value from Retry-After header (seconds or datetime string)

Returns:

Delay in seconds before next retry

nautical.io.retry.should_retry(error: Exception, config: RetryConfig) → tuple[bool, str | None][source]

Determine if an error should trigger a retry.

Parameters:

  • error – Exception that was raised

  • config – RetryConfig instance

Returns:

(should_retry: bool, retry_after: str|None)

Return type:

tuple

nautical.io.retry.with_retry(config: RetryConfig | None = None, rate_limiter: RateLimiter | None = None) → Callable[source]

Decorator to add retry logic to a function.

Parameters:

  • config – RetryConfig instance, or None for defaults

  • rate_limiter – RateLimiter instance, or None to use global limiter

Example

@with_retry(RetryConfig(max_retries=5)) def fetch_data(url):

return urlopen(url)

nautical.io.retry.retry_request(func: Callable, config: RetryConfig | None = None, rate_limiter: RateLimiter | None = None, *args: Any, **kwargs: Any) → Any[source]

Retry a function call with retry logic.

Alternative to decorator when you can’t use @with_retry.

Parameters:

  • func – Function to call

  • config – RetryConfig instance

  • rate_limiter – RateLimiter instance

  • *args – Arguments to pass to func

  • **kwargs –

    Arguments to pass to func

Returns:

Result of func(*args, **kwargs)

Example

result = retry_request(

urlopen, config=RetryConfig(max_retries=5), url=”https://example.com”

)

nautical.io.sources module

nautical.io.sources.get_buoy_sources(source_types: SourceType | list[SourceType] = SourceType.ALL) → dict[str, Source][source]

NOAA is kind enough to provide all of names, ids, and other information about ALL of their known buoys in a kml document hosted at the link provided (https://www.ndbc.noaa.gov/kml/marineobs_by_pgm.kml). Read through this document and parse the buoy information to determine their id and location. The ID can be used to provide to get_noaa_forecast_url(). Then we can find even more information about the buoys.

Returns:

dictionary all source names mapped to their respective source.

nautical.io.sources.validate_sources(source_data: dict[str, Source], remove_invalid: bool = True) → dict[str, Source][source]

This function is presumed to be executed after get_buoy_sources. The results of the previous function meet the requirements for the formatted parameter here. The function will attempt to parse all buoys found for each source supplied.

Parameters:

  • source_data – Dictionary in the format of source_name: source

  • remove_invalid – when True [default] remove the buoys that are invalid

Returns:

New dictionary where the buoys for each source are validated

nautical.io.web module

nautical.io.web.get_noaa_forecast_url(buoy: str) → str | None[source]

NOAA is kind enough to post all of their data from their buoys at the same url ONLY requiring the id of buoy to change at the end of the link (https://www.ndbc.noaa.gov/station_page.php?station=). This function will simply take in the buoy from the user and append the data to the end of the url, IFF the data exists.

Parameters:

buoy – id of the buoy

Returns:

full url if buoy is not empty, otherwise None

nautical.io.web.get_url_source(url_name: str) → BeautifulSoup[source]

Fetch and parse NOAA buoy data webpage with automatic retry and rate limiting.

The function automatically retries on network errors, timeouts, and server errors (5xx status codes). Rate limiting (HTTP 429) is detected and respected via the Retry-After header. Client errors (4xx except 429) are not retried.

Parameters:

url_name – URL to fetch

Returns:

BeautifulSoup object containing the parsed HTML

Raises:

  • HTTPError – If the request fails after all retries

  • URLError – If there’s a network-level error after all retries

  • ValueError – If url_name is invalid

Module contents

The io module consists of the functions utilized to search for information about buoys and their sources on NOAA’s website.

nautical.io.get_url_source(url_name: str) → BeautifulSoup[source]

Fetch and parse NOAA buoy data webpage with automatic retry and rate limiting.

The function automatically retries on network errors, timeouts, and server errors (5xx status codes). Rate limiting (HTTP 429) is detected and respected via the Retry-After header. Client errors (4xx except 429) are not retried.

Parameters:

url_name – URL to fetch

Returns:

BeautifulSoup object containing the parsed HTML

Raises:

  • HTTPError – If the request fails after all retries

  • URLError – If there’s a network-level error after all retries

  • ValueError – If url_name is invalid

nautical.io.get_noaa_forecast_url(buoy: str) → str | None[source]

NOAA is kind enough to post all of their data from their buoys at the same url ONLY requiring the id of buoy to change at the end of the link (https://www.ndbc.noaa.gov/station_page.php?station=). This function will simply take in the buoy from the user and append the data to the end of the url, IFF the data exists.

Parameters:

buoy – id of the buoy

Returns:

full url if buoy is not empty, otherwise None

nautical.io.create_buoy(buoy: str) → Buoy | None[source]

Provide a full workup for a specific buoy. If the buoy is None or it cannot be found then the data returned will be considered invalid as None

Parameters:

buoy – id of the buoy to do a workup on

Returns:

BuoyWorkup if successful else None

nautical.io.fill_buoy(buoy: Buoy) → None[source]

Pass in a Buoy object that needs to be filled in with the current data. The buoy object will have the validity set if the results were successful

Parameters:

buoy – nautical.noaa.buoy.Buoy object

nautical.io.get_current_data(soup: BeautifulSoup, buoy: BuoyData, search: str | list[str]) → bool[source]

Search the beautiful soup object for a TABLE containing the search string. The function will grab the data from the table and create a NOAAData object and return the data

Parameters:

  • soup – beautiful soup object generated from the get_url_source()

  • buoy – BuoyData object that should be filled with data as this function parses the data.

  • search – text to search for in the soup object.

Returns:

True when data has been found and set

nautical.io.get_buoy_data(soup: BeautifulSoup, buoy: BuoyData, search: str | list[str]) → bool

Search the beautiful soup object for a TABLE containing the search string. The function will grab the data from the table and create a NOAAData object and return the data

Parameters:

  • soup – beautiful soup object generated from the get_url_source()

  • buoy – BuoyData object that should be filled with data as this function parses the data.

  • search – text to search for in the soup object.

Returns:

True when data has been found and set

nautical.io.get_buoy_sources(source_types: SourceType | list[SourceType] = SourceType.ALL) → dict[str, Source][source]

NOAA is kind enough to provide all of names, ids, and other information about ALL of their known buoys in a kml document hosted at the link provided (https://www.ndbc.noaa.gov/kml/marineobs_by_pgm.kml). Read through this document and parse the buoy information to determine their id and location. The ID can be used to provide to get_noaa_forecast_url(). Then we can find even more information about the buoys.

Returns:

dictionary all source names mapped to their respective source.

nautical.io.parse_winds(wind_data: str) → dict[str, float | None][source]

Parse the wind information. The wind data includes a direction, speed in knots, as well as the gust speed

Parameters:

wind_data – String containing the wind string

Returns:

Dictionary containing the windspeed and gust information

nautical.io.parse_location(location_data: str) → dict[source]

Parse the latitude and longitude values out of the the string that was passed in. The latitude and longitude should contain the NSEW strings describing their sign.

Parameters:

location_data – String data contain latitude and longitude values

Returns:

dictionary containing the location point (when valid)

nautical.io.parse_time(time_data: str) → dict[source]

Parse the month/day/year time out of the time data string from CDATA.

Parameters:

time_data – String containing all time and date information

Returns:

dictionary mm, dd, year, time where time is NauticalTime

nautical.io.parse_cdata(cdata: str) → dict[source]

Parse the CDATA string that contains information about this presumed buoy. The data here will be used as supplemental information as some buoys cannot be scraped online (no valid or useable data).

Parameters:

cdata – String containing the CDATA or description element from kml

Returns:

Dictionary containing all parsed fields.

nautical.io.fill_buoy_with_cdata(buoy: Buoy, cdata: str) → None[source]

Parse the CDATA string that contains information about this presumed buoy. The data here will be used as supplemental information as some buoys cannot be scraped online (no valid or useable data).

Parameters:

  • buoy – nautical.noaa.buoy.Buoy

  • cdata – String containing the CDATA or description element from kml

Returns:

Buoy object