Documentation for aalibrary.tugboat_api.TugboatAPI

The TugboatAPI class provides a way to access the the API for Tugboat through AALibrary. There are functions for submitting datasets, and even functions for creating and updating metadata related to Tugboat.

This class provides methods to interact with the Tugboat API.

Methods:

Name	Description
__init__	Initializes the TugboatAPI class.
check_connection	Fetches a sea area by its ID from the Tugboat API to test the
create_empty_submission_file	Creates an empty submission file at the file_download_location.
get_all_instruments	Fetches all instruments from the Tugboat API.
get_all_jobs	Fetches all jobs from the Tugboat API. Useful for checking
get_all_organizations	Fetches all organizations from the Tugboat API.
get_all_people	Fetches all people & their info from the Tugboat API.
get_all_platforms	Fetches all platforms from the Tugboat API.
get_all_sea_areas	Fetches all sea areas from the Tugboat API.
get_all_submissions	Fetches all submissions from the Tugboat API.
get_instrument_by_id	Fetches an instrument by its ID from the Tugboat API.
get_job_by_id	Fetches a job by its ID from the Tugboat API. Useful for checking
get_organization_by_id	Fetches an organization by its ID from the Tugboat API.
get_person_by_id	Fetches a person by their ID from the Tugboat API.
get_platform_by_id	Fetches a platform by its ID from the Tugboat API.
get_sea_area_by_id	Fetches a sea area by its ID from the Tugboat API.
get_submission_by_id	Fetches a submission by its ID from the Tugboat API.
get_submission_job_by_id	Fetches the job associated with a submission by its ID from the
post_new_instrument	Posts a new instrument to the Tugboat API.
post_new_organization	Posts a new organization to the Tugboat API.
post_new_person	Posts a new person to the Tugboat API.
post_new_platform	Posts a new platform to the Tugboat API.
post_new_sea_area	Posts a new sea area to the Tugboat API.
post_new_submission	Posts a new submission to the Tugboat API.
resubmit_submission	Resubmits a submission to the Tugboat API.
search_instruments_by_short_name	Searches for instruments by short name in the Tugboat API.
search_people_by_email	Searches for people by email in the Tugboat API.
search_people_by_name	Searches for people by name in the Tugboat API.
update_person_by_id	Updates a person by their ID in the Tugboat API.
validate_submission_json	Validates a submission JSON (supplied through a file path or dict)

Source code in src\aalibrary\tugboat_api.py

class TugboatAPI:
    """
    This class provides methods to interact with the Tugboat API.
    """

    # TODO: Make it so that the project id and bucket name can be passed
    # in as arguments to the class init function.

    project_id: str = "ggn-nmfs-aa-dev-1"
    gcp_bucket_name: str = "ggn-nmfs-aa-dev-1-data"
    empty_submission_file_path: str = (
        "other/tugboat_empty_submission_template.json"
    )
    __tugboat_cred: str = None
    tugboat_api_url: str = (
        "https://nih-uat-tugboat.nesdis-hq.noaa.gov:5443/api/v1/"
    )
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
    }

    def __init__(self, **kwargs):
        """
        Initializes the TugboatAPI class.
        """
        self.__dict__.update(kwargs)
        self._create_vars()
        self._set_tugboat_credentials()

    def _create_vars(self):
        """Creates vars for use later."""

        # Create google connection objects to download the Tugboat creds
        # if (
        #     ("gcp_stor_client" not in self.__dict__)
        #     or ("gcp_bucket_name" not in self.__dict__)
        #     or ("gcp_bucket" not in self.__dict__)
        # ):
        #     self.gcp_stor_client, self.gcp_bucket_name, self.gcp_bucket = (
        #         cloud_utils.setup_gcp_storage_objs(
        #             project_id=self.project_id,
        #             gcp_bucket_name=self.gcp_bucket_name,
        #         )
        #     )

    def _set_tugboat_credentials(self):
        """Sets the tugboat credentials for AALibrary from the google storage
        bucket as an environment var."""

        if self.__tugboat_cred is None:
            # If there is a dot-env file in the current directory, load it.
            if os.path.exists(".env"):
                load_dotenv()
            else:
                # Load from GCP bucket
                # Download as string so that we dont have to worry about
                # storing it anywhere
                # os.environ["TUGBOAT_CREDENTIALS"] = (
                #     cloud_utils.download_file_from_gcp_as_string(
                #         gcp_bucket=self.gcp_bucket,
                #         blob_file_path="other/tugboat_creds",
                #     )
                # )
                ...
            # Set the credentials
            self.__tugboat_cred = os.environ.get("TUGBOAT_CREDENTIALS")
        self.headers["Authorization"] = f"Bearer {self.__tugboat_cred}"

    def _get_request_as_json(self, url: str) -> dict:
        """Helper function to make a GET request and return the response as
        JSON."""

        try:
            response = requests.get(
                url, headers=self.headers, timeout=10, verify=False
            )
            # Raise an HTTPError for bad responses(4xx or 5xx)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error during GET request: {e}")
            raise requests.exceptions.RequestException

    def _post_request_as_json(self, url: str, payload: dict) -> dict:
        """Helper function to make a POST request and return the response as
        JSON.

        Args:
            url (str): The URL to make the POST request to.
            payload (dict): The JSON payload for the POST request.

        Returns:
            dict: The response from the POST request as a JSON dictionary."""
        try:
            response = requests.post(
                url,
                headers=self.headers,
                json=payload,
                timeout=10,
                verify=False,
            )
            # Raise an HTTPError for bad responses(4xx or 5xx)
            response.raise_for_status()
            print("POST request successful!")
            print("Response JSON:")
            print(response.json())
            return response.json()
        except Exception as e:
            print(
                f"POST request failed with status code: {response.status_code}"
            )
            print(response.text)
            raise requests.exceptions.RequestException from e

    def _put_request_as_json(self, url: str, payload: dict) -> dict:
        """Helper function to make a PUT request and return the response as
        JSON.

        Args:
            url (str): The URL to make the PUT request to.
            payload (dict): The JSON payload for the PUT request.

        Returns:
            dict: The response from the PUT request as a JSON dictionary."""

        try:
            response = requests.put(
                url,
                headers=self.headers,
                json=payload,
                timeout=10,
                verify=False,
            )
            # Raise an HTTPError for bad responses(4xx or 5xx)
            response.raise_for_status()
            print("PUT request successful!")
            print("Response JSON:")
            print(response.json())
            return response.json()
        except Exception as e:
            print(
                f"PUT request failed with status code: {response.status_code}"
            )
            print(response.text)
            raise requests.exceptions.RequestException from e

    ############################################################### CONNECTION
    def check_connection(self) -> bool:
        """Fetches a sea area by its ID from the Tugboat API to test the
        connection.

        Returns:
            bool: True if the connection is successful, False otherwise.
        """
        # Fetch the sea area with the specified ID (using 51 as an example)
        url = urllib.parse.urljoin(self.tugboat_api_url, "sea-areas/51")
        try:
            self._get_request_as_json(url)
            return True
        except requests.exceptions.RequestException:
            return False

    ############################################################### SUBMISSIONS
    def create_empty_submission_file(
        self, file_download_directory: str = ".", file_name: str = ""
    ):
        """Creates an empty submission file at the file_download_location.

        Args:
            file_download_directory (str, optional): The directory where you
                want to download the file. Defaults to ".".
            file_name (str, optional): _description_. Defaults to "".
        """

        # Normalize paths
        file_download_directory = (
            os.path.normpath(file_download_directory) + os.sep
        )
        # Convert locations into directories as needed.
        if file_download_directory != ".":
            # Edge-case: when dirname is passed ".", it responds with ""
            file_download_directory = (
                os.path.dirname(file_download_directory) + os.sep
            )
        file_download_path = os.path.normpath(
            os.sep.join([file_download_directory, file_name])
        )

        # Read empty submission file
        with open(self.empty_submission_file_path, "r", encoding="utf-8") as f:
            empty_submission = json.load(f)

        # Write empty submission file to the specified path
        with open(file_download_path, "w", encoding="utf-8") as f:
            json.dump(empty_submission, f, indent=4)

        print(f"Empty submission file created at: {file_download_path}")
        print("Please update the file with your submission details.")
        return file_download_path

    def validate_submission_json(
        self, submission_json_file_path: str = "", submission_dict: dict = None
    ):
        """Validates a submission JSON (supplied through a file path or dict)
        using the TugboatValidator. Prints out the errors in the submission to
        the console stderr.

        Args:
            submission_json_file_path (str, optional): The file path to the
                submission JSON. Defaults to "".
            submission_dict (dict, optional): The dictionary containing the
                submission details. Defaults to None.
        """

        # validator = TugboatValidator(
        #     submission_json_file_path=submission_json_file_path,
        #     submission_dict=submission_dict,
        # )
        # validator.validate_submission()

    def post_new_submission(self, submission_json_file_path: str = ""):
        """Posts a new submission to the Tugboat API."""

        # Create the URL for the submission endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "submissions")
        # Read the submission JSON file
        with open(submission_json_file_path, "r", encoding="utf-8") as f:
            submission_payload = json.load(f)

        return self._post_request_as_json(url, submission_payload)

    def resubmit_submission(
        self, submission_id: str, submission_json_file_path: str = ""
    ):
        """Resubmits a submission to the Tugboat API."""
        # Create the URL for the resubmission endpoint
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"submissions/resubmit/{submission_id}"
        )
        # Read the submission JSON file
        with open(submission_json_file_path, "r", encoding="utf-8") as f:
            submission_payload = json.load(f)
        response = requests.post(
            url,
            headers=self.headers,
            json=submission_payload,
            timeout=10,
            verify=False,
        )
        # Checking the response status code
        if response.status_code == 201:  # 201 Created for successful POST
            print("POST request successful!")
            print("Response JSON:")
            print(response.json())
        else:
            print(
                f"POST request failed with status code: {response.status_code}"
            )
            print(response.text)

    def get_all_submissions(self):
        """Fetches all submissions from the Tugboat API."""
        page = 1
        all_submissions = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"submissions?page={page}"
            )
            submissions = self._get_request_as_json(url)["items"]
            if submissions == []:
                break
            all_submissions.extend(submissions)
            page += 1
        return all_submissions

    def get_submission_by_id(self, submission_id: str):
        """Fetches a submission by its ID from the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"submissions/{submission_id}"
        )
        return self._get_request_as_json(url)

    def get_submission_job_by_id(self, submission_id: str):
        """Fetches the job associated with a submission by its ID from the
        Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"submissions/{submission_id}/job"
        )
        return self._get_request_as_json(url)

    ###################################################################### JOBS
    def get_all_jobs(self):
        """Fetches all jobs from the Tugboat API. Useful for checking
        the status of a submission."""

        page = 1
        all_jobs = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"jobs?page={page}"
            )
            jobs = self._get_request_as_json(url)["items"]
            if jobs == []:
                break
            all_jobs.extend(jobs)
            page += 1
        return all_jobs

    def get_job_by_id(self, job_id: str):
        """Fetches a job by its ID from the Tugboat API. Useful for checking
        the status of a submission."""

        url = urllib.parse.urljoin(self.tugboat_api_url, f"jobs/{job_id}")
        return self._get_request_as_json(url)

    #################################################################### PEOPLE
    def get_all_people(self) -> list:
        """Fetches all people & their info from the Tugboat API."""

        page = 1
        all_people = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"people?page={page}"
            )
            people = self._get_request_as_json(url)["items"]
            if people == []:
                break
            all_people.extend(people)
            page += 1
        return all_people

    def get_person_by_id(self, person_id: str) -> dict:
        """Fetches a person by their ID from the Tugboat API."""
        url = urllib.parse.urljoin(self.tugboat_api_url, f"people/{person_id}")
        return self._get_request_as_json(url)

    def update_person_by_id(self, person_id: str, person_json: dict) -> dict:
        """Updates a person by their ID in the Tugboat API.

        Args:
            person_id (str): The id of the person you are trying to update.
            person_json (dict): The JSON payload containing the updated person
                information.
                Example:
                    {
                        "name": "John Smith",
                        "position": "Water Column Data Manager",
                        "organization": "NESDIS;NCEI",
                        "street": "NOAA;NCEI, 325 Broadway",
                        "city": "Boulder",
                        "state": "CO",
                        "zip": "80305",
                        "country": "USA"
                    }

        Returns:
            dict: The response from the Tugboat API after updating the person.
        """
        url = urllib.parse.urljoin(self.tugboat_api_url, f"people/{person_id}")

        return self._put_request_as_json(url, person_json)

    def search_people_by_email(self, email: str) -> dict:
        """Searches for people by email in the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url,
            f"people?email={urllib.parse.quote(email)}",
        )
        resp = self._get_request_as_json(url)
        if resp["totalItems"] == 0:
            return None
        else:
            return resp["items"]

    def search_people_by_name(self, name: str) -> dict:
        """Searches for people by name in the Tugboat API.
        NOTE: Make sure the name matches exactly (case-sensitive) with the name
        in the Tugboat database, otherwise it will not return the person."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url,
            f"people?name={urllib.parse.quote(name)}",
        )
        resp = self._get_request_as_json(url)
        if resp["totalItems"] == 0:
            return None
        else:
            return resp["items"]

    def post_new_person(self, person_json: dict) -> dict:
        """Posts a new person to the Tugboat API.

        Args:
            person_json (dict): The dictionary containing the person's
                information.
                Example:
                    {
                        "name": "John Smith",
                        "position": "Water Column Data Manager",
                        "organization": "NESDIS;NCEI",
                        "street": "NOAA;NCEI, 325 Broadway",
                        "city": "Boulder",
                        "state": "CO",
                        "zip": "80305",
                        "country": "USA"
                    }

        Returns:
            dict: The response from the Tugboat API after creating the person.
                This includes the ID of the newly created person, which can be
                used for future reference.
        """

        # Create the URL for the person endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "people")

        return self._post_request_as_json(url, person_json)

    ############################################################# ORGANIZATIONS
    def get_organization_by_id(self, organization_id: str) -> dict:
        """Fetches an organization by its ID from the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"organizations/{organization_id}"
        )
        return self._get_request_as_json(url)

    def get_all_organizations(self) -> list:
        """Fetches all organizations from the Tugboat API."""
        page = 1
        all_organizations = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"organizations?page={page}"
            )
            organizations = self._get_request_as_json(url)["items"]
            if organizations == []:
                break
            all_organizations.extend(organizations)
            page += 1
        return all_organizations

    def post_new_organization(self, organization_json: dict) -> dict:
        """Posts a new organization to the Tugboat API.

        Args:
            organization_json (dict): The dictionary containing the
                organization's information.
                Example:
                    {
                        "name": "NOAA AFSC",
                        "street": "7600 Sand Point Way N.E., Building 4",
                        "city": "Seattle",
                        "state": "WA",
                        "zip": "98115",
                        "country": "USA",
                        "phone": "206-526-4000"
                    }

        Returns:
            dict: The response from the Tugboat API after creating the
                organization. This includes the ID of the newly created
                organization, which can be used for future reference.
        """

        # Create the URL for the organization endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "organizations")

        return self._post_request_as_json(url, organization_json)

    ################################################################# PLATFORMS
    def get_platform_by_id(self, platform_id: str) -> dict:
        """Fetches a platform by its ID from the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"platforms/{platform_id}"
        )
        return self._get_request_as_json(url)

    def get_all_platforms(self) -> list:
        """Fetches all platforms from the Tugboat API."""
        page = 1
        all_platforms = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"platforms?page={page}"
            )
            platforms = self._get_request_as_json(url)["items"]
            if platforms == []:
                break
            all_platforms.extend(platforms)
            page += 1
        return all_platforms

    def post_new_platform(self, platform_json: dict) -> dict:
        """Posts a new platform to the Tugboat API.

        Args:
            platform_json (dict): The dictionary containing the
                platform's information.
                Example:
                    {
                        "name": "Towed array",
                    }

        Returns:
            dict: The response from the Tugboat API after creating the
                platform. This includes the ID of the newly created
                platform, which can be used for future reference.
        """

        # Create the URL for the platform endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "platforms")

        return self._post_request_as_json(url, platform_json)

    ############################################################### INSTRUMENTS
    def get_instrument_by_id(self, instrument_id: str) -> dict:
        """Fetches an instrument by its ID from the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"instruments/{instrument_id}"
        )
        return self._get_request_as_json(url)

    def search_instruments_by_short_name(self, short_name: str) -> dict:
        """Searches for instruments by short name in the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url,
            f"instruments?shortName={urllib.parse.quote(short_name)}",
        )
        resp = self._get_request_as_json(url)
        if resp["totalItems"] == 0:
            return None
        else:
            return resp["items"]

    def get_all_instruments(self) -> list:
        """Fetches all instruments from the Tugboat API."""
        page = 1
        all_instruments = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"instruments?page={page}"
            )
            instruments = self._get_request_as_json(url)["items"]
            if instruments == []:
                break
            all_instruments.extend(instruments)
            page += 1
        return all_instruments

    def post_new_instrument(self, instrument_json: dict) -> dict:
        """Posts a new instrument to the Tugboat API.

        Args:
            instrument_json (dict): The dictionary containing the instrument's
                information.
                Example:
                    {
                        "instrument": "Kongsberg EM122",
                        "shortName": "EM122",
                        "extensions": "[wcd, kmwcd]"
                    }

        Returns:
            dict: The response from the Tugboat API after creating the
                instrument. This includes the ID of the newly created
                instrument, which can be used for future reference.
        """

        # Create the URL for the instrument endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "instruments")

        return self._post_request_as_json(url, instrument_json)

    ################################################################# SEA AREAS
    def get_sea_area_by_id(self, sea_area_id: str) -> dict:
        """Fetches a sea area by its ID from the Tugboat API."""
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"sea-areas/{sea_area_id}"
        )
        return self._get_request_as_json(url)

    def get_all_sea_areas(self) -> list:
        """Fetches all sea areas from the Tugboat API."""
        page = 1
        all_sea_areas = []
        while True:
            url = urllib.parse.urljoin(
                self.tugboat_api_url, f"sea-areas?page={page}"
            )
            sea_areas = self._get_request_as_json(url)["items"]
            if sea_areas == []:
                break
            all_sea_areas.extend(sea_areas)
            page += 1
        return all_sea_areas

    def post_new_sea_area(self, sea_area_json: dict) -> dict:
        """Posts a new sea area to the Tugboat API.

        Args:
            sea_area_json (dict): The dictionary containing the sea area's
                information.
                Example:
                    {
                        "name": "North Pacific Ocean"
                    }

        Returns:
            dict: The response from the Tugboat API after creating the
                sea area. This includes the ID of the newly created
                sea area, which can be used for future reference.
        """

        # Create the URL for the sea area endpoint
        url = urllib.parse.urljoin(self.tugboat_api_url, "sea-areas")

        return self._post_request_as_json(url, sea_area_json)

__init__(**kwargs)

Initializes the TugboatAPI class.

Source code in src\aalibrary\tugboat_api.py

def __init__(self, **kwargs):
    """
    Initializes the TugboatAPI class.
    """
    self.__dict__.update(kwargs)
    self._create_vars()
    self._set_tugboat_credentials()

check_connection()

Fetches a sea area by its ID from the Tugboat API to test the connection.

Returns:

Name	Type	Description
bool	bool	True if the connection is successful, False otherwise.

Source code in src\aalibrary\tugboat_api.py

def check_connection(self) -> bool:
    """Fetches a sea area by its ID from the Tugboat API to test the
    connection.

    Returns:
        bool: True if the connection is successful, False otherwise.
    """
    # Fetch the sea area with the specified ID (using 51 as an example)
    url = urllib.parse.urljoin(self.tugboat_api_url, "sea-areas/51")
    try:
        self._get_request_as_json(url)
        return True
    except requests.exceptions.RequestException:
        return False

create_empty_submission_file(file_download_directory='.', file_name='')

Creates an empty submission file at the file_download_location.

Parameters:

Name	Type	Description	Default
file_download_directory	str	The directory where you want to download the file. Defaults to ".".	'.'
file_name	str	description. Defaults to "".	''

Source code in src\aalibrary\tugboat_api.py

def create_empty_submission_file(
    self, file_download_directory: str = ".", file_name: str = ""
):
    """Creates an empty submission file at the file_download_location.

    Args:
        file_download_directory (str, optional): The directory where you
            want to download the file. Defaults to ".".
        file_name (str, optional): _description_. Defaults to "".
    """

    # Normalize paths
    file_download_directory = (
        os.path.normpath(file_download_directory) + os.sep
    )
    # Convert locations into directories as needed.
    if file_download_directory != ".":
        # Edge-case: when dirname is passed ".", it responds with ""
        file_download_directory = (
            os.path.dirname(file_download_directory) + os.sep
        )
    file_download_path = os.path.normpath(
        os.sep.join([file_download_directory, file_name])
    )

    # Read empty submission file
    with open(self.empty_submission_file_path, "r", encoding="utf-8") as f:
        empty_submission = json.load(f)

    # Write empty submission file to the specified path
    with open(file_download_path, "w", encoding="utf-8") as f:
        json.dump(empty_submission, f, indent=4)

    print(f"Empty submission file created at: {file_download_path}")
    print("Please update the file with your submission details.")
    return file_download_path

get_all_instruments()

Fetches all instruments from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_instruments(self) -> list:
    """Fetches all instruments from the Tugboat API."""
    page = 1
    all_instruments = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"instruments?page={page}"
        )
        instruments = self._get_request_as_json(url)["items"]
        if instruments == []:
            break
        all_instruments.extend(instruments)
        page += 1
    return all_instruments

get_all_jobs()

Fetches all jobs from the Tugboat API. Useful for checking the status of a submission.

Source code in src\aalibrary\tugboat_api.py

def get_all_jobs(self):
    """Fetches all jobs from the Tugboat API. Useful for checking
    the status of a submission."""

    page = 1
    all_jobs = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"jobs?page={page}"
        )
        jobs = self._get_request_as_json(url)["items"]
        if jobs == []:
            break
        all_jobs.extend(jobs)
        page += 1
    return all_jobs

get_all_organizations()

Fetches all organizations from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_organizations(self) -> list:
    """Fetches all organizations from the Tugboat API."""
    page = 1
    all_organizations = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"organizations?page={page}"
        )
        organizations = self._get_request_as_json(url)["items"]
        if organizations == []:
            break
        all_organizations.extend(organizations)
        page += 1
    return all_organizations

get_all_people()

Fetches all people & their info from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_people(self) -> list:
    """Fetches all people & their info from the Tugboat API."""

    page = 1
    all_people = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"people?page={page}"
        )
        people = self._get_request_as_json(url)["items"]
        if people == []:
            break
        all_people.extend(people)
        page += 1
    return all_people

get_all_platforms()

Fetches all platforms from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_platforms(self) -> list:
    """Fetches all platforms from the Tugboat API."""
    page = 1
    all_platforms = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"platforms?page={page}"
        )
        platforms = self._get_request_as_json(url)["items"]
        if platforms == []:
            break
        all_platforms.extend(platforms)
        page += 1
    return all_platforms

get_all_sea_areas()

Fetches all sea areas from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_sea_areas(self) -> list:
    """Fetches all sea areas from the Tugboat API."""
    page = 1
    all_sea_areas = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"sea-areas?page={page}"
        )
        sea_areas = self._get_request_as_json(url)["items"]
        if sea_areas == []:
            break
        all_sea_areas.extend(sea_areas)
        page += 1
    return all_sea_areas

get_all_submissions()

Fetches all submissions from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_all_submissions(self):
    """Fetches all submissions from the Tugboat API."""
    page = 1
    all_submissions = []
    while True:
        url = urllib.parse.urljoin(
            self.tugboat_api_url, f"submissions?page={page}"
        )
        submissions = self._get_request_as_json(url)["items"]
        if submissions == []:
            break
        all_submissions.extend(submissions)
        page += 1
    return all_submissions

get_instrument_by_id(instrument_id)

Fetches an instrument by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_instrument_by_id(self, instrument_id: str) -> dict:
    """Fetches an instrument by its ID from the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"instruments/{instrument_id}"
    )
    return self._get_request_as_json(url)

get_job_by_id(job_id)

Fetches a job by its ID from the Tugboat API. Useful for checking the status of a submission.

Source code in src\aalibrary\tugboat_api.py

def get_job_by_id(self, job_id: str):
    """Fetches a job by its ID from the Tugboat API. Useful for checking
    the status of a submission."""

    url = urllib.parse.urljoin(self.tugboat_api_url, f"jobs/{job_id}")
    return self._get_request_as_json(url)

get_organization_by_id(organization_id)

Fetches an organization by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_organization_by_id(self, organization_id: str) -> dict:
    """Fetches an organization by its ID from the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"organizations/{organization_id}"
    )
    return self._get_request_as_json(url)

get_person_by_id(person_id)

Fetches a person by their ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_person_by_id(self, person_id: str) -> dict:
    """Fetches a person by their ID from the Tugboat API."""
    url = urllib.parse.urljoin(self.tugboat_api_url, f"people/{person_id}")
    return self._get_request_as_json(url)

get_platform_by_id(platform_id)

Fetches a platform by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_platform_by_id(self, platform_id: str) -> dict:
    """Fetches a platform by its ID from the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"platforms/{platform_id}"
    )
    return self._get_request_as_json(url)

get_sea_area_by_id(sea_area_id)

Fetches a sea area by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_sea_area_by_id(self, sea_area_id: str) -> dict:
    """Fetches a sea area by its ID from the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"sea-areas/{sea_area_id}"
    )
    return self._get_request_as_json(url)

get_submission_by_id(submission_id)

Fetches a submission by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_submission_by_id(self, submission_id: str):
    """Fetches a submission by its ID from the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"submissions/{submission_id}"
    )
    return self._get_request_as_json(url)

get_submission_job_by_id(submission_id)

Fetches the job associated with a submission by its ID from the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def get_submission_job_by_id(self, submission_id: str):
    """Fetches the job associated with a submission by its ID from the
    Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"submissions/{submission_id}/job"
    )
    return self._get_request_as_json(url)

post_new_instrument(instrument_json)

Posts a new instrument to the Tugboat API.

Parameters:

Name	Type	Description	Default
instrument_json	dict	The dictionary containing the instrument's information. Example: { "instrument": "Kongsberg EM122", "shortName": "EM122", "extensions": "[wcd, kmwcd]" }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after creating the instrument. This includes the ID of the newly created instrument, which can be used for future reference.

Source code in src\aalibrary\tugboat_api.py

def post_new_instrument(self, instrument_json: dict) -> dict:
    """Posts a new instrument to the Tugboat API.

    Args:
        instrument_json (dict): The dictionary containing the instrument's
            information.
            Example:
                {
                    "instrument": "Kongsberg EM122",
                    "shortName": "EM122",
                    "extensions": "[wcd, kmwcd]"
                }

    Returns:
        dict: The response from the Tugboat API after creating the
            instrument. This includes the ID of the newly created
            instrument, which can be used for future reference.
    """

    # Create the URL for the instrument endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "instruments")

    return self._post_request_as_json(url, instrument_json)

post_new_organization(organization_json)

Posts a new organization to the Tugboat API.

Parameters:

Name	Type	Description	Default
organization_json	dict	The dictionary containing the organization's information. Example: { "name": "NOAA AFSC", "street": "7600 Sand Point Way N.E., Building 4", "city": "Seattle", "state": "WA", "zip": "98115", "country": "USA", "phone": "206-526-4000" }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after creating the organization. This includes the ID of the newly created organization, which can be used for future reference.

Source code in src\aalibrary\tugboat_api.py

def post_new_organization(self, organization_json: dict) -> dict:
    """Posts a new organization to the Tugboat API.

    Args:
        organization_json (dict): The dictionary containing the
            organization's information.
            Example:
                {
                    "name": "NOAA AFSC",
                    "street": "7600 Sand Point Way N.E., Building 4",
                    "city": "Seattle",
                    "state": "WA",
                    "zip": "98115",
                    "country": "USA",
                    "phone": "206-526-4000"
                }

    Returns:
        dict: The response from the Tugboat API after creating the
            organization. This includes the ID of the newly created
            organization, which can be used for future reference.
    """

    # Create the URL for the organization endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "organizations")

    return self._post_request_as_json(url, organization_json)

post_new_person(person_json)

Posts a new person to the Tugboat API.

Parameters:

Name	Type	Description	Default
person_json	dict	The dictionary containing the person's information. Example: { "name": "John Smith", "position": "Water Column Data Manager", "organization": "NESDIS;NCEI", "street": "NOAA;NCEI, 325 Broadway", "city": "Boulder", "state": "CO", "zip": "80305", "country": "USA" }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after creating the person. This includes the ID of the newly created person, which can be used for future reference.

Source code in src\aalibrary\tugboat_api.py

def post_new_person(self, person_json: dict) -> dict:
    """Posts a new person to the Tugboat API.

    Args:
        person_json (dict): The dictionary containing the person's
            information.
            Example:
                {
                    "name": "John Smith",
                    "position": "Water Column Data Manager",
                    "organization": "NESDIS;NCEI",
                    "street": "NOAA;NCEI, 325 Broadway",
                    "city": "Boulder",
                    "state": "CO",
                    "zip": "80305",
                    "country": "USA"
                }

    Returns:
        dict: The response from the Tugboat API after creating the person.
            This includes the ID of the newly created person, which can be
            used for future reference.
    """

    # Create the URL for the person endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "people")

    return self._post_request_as_json(url, person_json)

post_new_platform(platform_json)

Posts a new platform to the Tugboat API.

Parameters:

Name	Type	Description	Default
platform_json	dict	The dictionary containing the platform's information. Example: { "name": "Towed array", }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after creating the platform. This includes the ID of the newly created platform, which can be used for future reference.

Source code in src\aalibrary\tugboat_api.py

def post_new_platform(self, platform_json: dict) -> dict:
    """Posts a new platform to the Tugboat API.

    Args:
        platform_json (dict): The dictionary containing the
            platform's information.
            Example:
                {
                    "name": "Towed array",
                }

    Returns:
        dict: The response from the Tugboat API after creating the
            platform. This includes the ID of the newly created
            platform, which can be used for future reference.
    """

    # Create the URL for the platform endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "platforms")

    return self._post_request_as_json(url, platform_json)

post_new_sea_area(sea_area_json)

Posts a new sea area to the Tugboat API.

Parameters:

Name	Type	Description	Default
sea_area_json	dict	The dictionary containing the sea area's information. Example: { "name": "North Pacific Ocean" }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after creating the sea area. This includes the ID of the newly created sea area, which can be used for future reference.

Source code in src\aalibrary\tugboat_api.py

def post_new_sea_area(self, sea_area_json: dict) -> dict:
    """Posts a new sea area to the Tugboat API.

    Args:
        sea_area_json (dict): The dictionary containing the sea area's
            information.
            Example:
                {
                    "name": "North Pacific Ocean"
                }

    Returns:
        dict: The response from the Tugboat API after creating the
            sea area. This includes the ID of the newly created
            sea area, which can be used for future reference.
    """

    # Create the URL for the sea area endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "sea-areas")

    return self._post_request_as_json(url, sea_area_json)

post_new_submission(submission_json_file_path='')

Posts a new submission to the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def post_new_submission(self, submission_json_file_path: str = ""):
    """Posts a new submission to the Tugboat API."""

    # Create the URL for the submission endpoint
    url = urllib.parse.urljoin(self.tugboat_api_url, "submissions")
    # Read the submission JSON file
    with open(submission_json_file_path, "r", encoding="utf-8") as f:
        submission_payload = json.load(f)

    return self._post_request_as_json(url, submission_payload)

resubmit_submission(submission_id, submission_json_file_path='')

Resubmits a submission to the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def resubmit_submission(
    self, submission_id: str, submission_json_file_path: str = ""
):
    """Resubmits a submission to the Tugboat API."""
    # Create the URL for the resubmission endpoint
    url = urllib.parse.urljoin(
        self.tugboat_api_url, f"submissions/resubmit/{submission_id}"
    )
    # Read the submission JSON file
    with open(submission_json_file_path, "r", encoding="utf-8") as f:
        submission_payload = json.load(f)
    response = requests.post(
        url,
        headers=self.headers,
        json=submission_payload,
        timeout=10,
        verify=False,
    )
    # Checking the response status code
    if response.status_code == 201:  # 201 Created for successful POST
        print("POST request successful!")
        print("Response JSON:")
        print(response.json())
    else:
        print(
            f"POST request failed with status code: {response.status_code}"
        )
        print(response.text)

search_instruments_by_short_name(short_name)

Searches for instruments by short name in the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def search_instruments_by_short_name(self, short_name: str) -> dict:
    """Searches for instruments by short name in the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url,
        f"instruments?shortName={urllib.parse.quote(short_name)}",
    )
    resp = self._get_request_as_json(url)
    if resp["totalItems"] == 0:
        return None
    else:
        return resp["items"]

search_people_by_email(email)

Searches for people by email in the Tugboat API.

Source code in src\aalibrary\tugboat_api.py

def search_people_by_email(self, email: str) -> dict:
    """Searches for people by email in the Tugboat API."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url,
        f"people?email={urllib.parse.quote(email)}",
    )
    resp = self._get_request_as_json(url)
    if resp["totalItems"] == 0:
        return None
    else:
        return resp["items"]

search_people_by_name(name)

Searches for people by name in the Tugboat API. NOTE: Make sure the name matches exactly (case-sensitive) with the name in the Tugboat database, otherwise it will not return the person.

Source code in src\aalibrary\tugboat_api.py

def search_people_by_name(self, name: str) -> dict:
    """Searches for people by name in the Tugboat API.
    NOTE: Make sure the name matches exactly (case-sensitive) with the name
    in the Tugboat database, otherwise it will not return the person."""
    url = urllib.parse.urljoin(
        self.tugboat_api_url,
        f"people?name={urllib.parse.quote(name)}",
    )
    resp = self._get_request_as_json(url)
    if resp["totalItems"] == 0:
        return None
    else:
        return resp["items"]

update_person_by_id(person_id, person_json)

Updates a person by their ID in the Tugboat API.

Parameters:

Name	Type	Description	Default
person_id	str	The id of the person you are trying to update.	required
person_json	dict	The JSON payload containing the updated person information. Example: { "name": "John Smith", "position": "Water Column Data Manager", "organization": "NESDIS;NCEI", "street": "NOAA;NCEI, 325 Broadway", "city": "Boulder", "state": "CO", "zip": "80305", "country": "USA" }	required

Returns:

Name	Type	Description
dict	dict	The response from the Tugboat API after updating the person.

Source code in src\aalibrary\tugboat_api.py

def update_person_by_id(self, person_id: str, person_json: dict) -> dict:
    """Updates a person by their ID in the Tugboat API.

    Args:
        person_id (str): The id of the person you are trying to update.
        person_json (dict): The JSON payload containing the updated person
            information.
            Example:
                {
                    "name": "John Smith",
                    "position": "Water Column Data Manager",
                    "organization": "NESDIS;NCEI",
                    "street": "NOAA;NCEI, 325 Broadway",
                    "city": "Boulder",
                    "state": "CO",
                    "zip": "80305",
                    "country": "USA"
                }

    Returns:
        dict: The response from the Tugboat API after updating the person.
    """
    url = urllib.parse.urljoin(self.tugboat_api_url, f"people/{person_id}")

    return self._put_request_as_json(url, person_json)

validate_submission_json(submission_json_file_path='', submission_dict=None)

Validates a submission JSON (supplied through a file path or dict) using the TugboatValidator. Prints out the errors in the submission to the console stderr.

Parameters:

Name	Type	Description	Default
submission_json_file_path	str	The file path to the submission JSON. Defaults to "".	''
submission_dict	dict	The dictionary containing the submission details. Defaults to None.	None

Source code in src\aalibrary\tugboat_api.py

def validate_submission_json(
    self, submission_json_file_path: str = "", submission_dict: dict = None
):
    """Validates a submission JSON (supplied through a file path or dict)
    using the TugboatValidator. Prints out the errors in the submission to
    the console stderr.

    Args:
        submission_json_file_path (str, optional): The file path to the
            submission JSON. Defaults to "".
        submission_dict (dict, optional): The dictionary containing the
            submission details. Defaults to None.
    """