Skip to content

Package Documentation

Signal's Port Congestion API Wrapper.

PortCongestionAPI

Signal's Port Congestion API Wrapper.

Source code in signal_ocean/port_congestion/port_congestion_api.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 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
class PortCongestionAPI:
    """Signal's Port Congestion API Wrapper."""

    base_url = "port-congestion-api"

    def __init__(self, connection: Optional[Connection] = None):
        """Initializes PortCongestionAPI.

        Args:
            connection (Connection, optional): API connection configuration.
                                               If not provided, the default
                                               connection method is used.
        """
        self.__connection = connection or Connection()

    def query_port_congestion(
        self,
        ports: Optional[List[str]] = None,
        port_ids: Optional[List[int]] = None,
        level_0_areas: Optional[List[str]] = None,
        level_0_area_ids: Optional[List[int]] = None,
        level_1_areas: Optional[List[str]] = None,
        level_1_area_ids: Optional[List[int]] = None,
        countries: Optional[List[str]] = None,
        country_ids: Optional[List[int]] = None,
        vessel_types: Optional[List[str]] = None,
        vessel_type_ids: Optional[List[int]] = None,
        vessel_classes: Optional[List[str]] = None,
        vessel_class_ids: Optional[List[int]] = None,
        date_from: Optional[date] = None,
    ) -> Optional[List[PortCongestionTimeSeriesEntry]]:
        """Exposes Port Congestion's `query` endpoint.

        Args:
            ports (list, optional): A list of Port Names. Defaults to None.
            port_ids (list, optional): A list of Port IDs. Defaults to None.
            level_0_areas (list, optional): A list of Level 0 Area names.
                                            Defaults to None.
            level_0_area_ids (list, optional): A list of Level 0 Area IDs.
                                               Defaults to None.
            level_1_areas (list, optional): A list of Level 1 Area names.
                                            Defaults to None.
            level_1_area_ids (list, optional): A list of Level 1 Area IDs.
                                               Defaults to None.
            countries (list, optional): A list of Country names.
                                        Defaults to None.
            country_ids (list, optional): A list of Country IDs.
                                          Defaults to None.
            vessel_types (list, optional): A list of Vessel Type names.
                                           Defaults to None.
            vessel_type_ids (list, optional): A list of Vessel Type IDs.
                                              Defaults to None.
            vessel_classes (list, optional): A list of Vessel Class names.
                                             Defaults to None.
            vessel_class_ids (list, optional): A list of Vessel Class IDs.
                                               Defaults to None.
            date_from (date, optional): A Date point to query from.
                                        Defaults to None.

        Raises:
            RuntimeError: In case of any request issue.

        Returns:
            List[PortCongestionTimeSeriesEntry], optional:
                A list of PortCongestionTimeSeriesEntry or None.
        """
        query_url = f"{self.base_url}/api/v1/query/"
        params = {
            "Ports": ports,
            "PortIDs": port_ids,
            "Level0Areas": level_0_areas,
            "Level0AreasIDs": level_0_area_ids,
            "Level1Areas": level_1_areas,
            "Level1AreaIDs": level_1_area_ids,
            "Countries": countries,
            "CountriesIDs": country_ids,
            "VesselTypes": vessel_types,
            "VesselTypeIDs": vessel_type_ids,
            "VesselClasses": vessel_classes,
            "VesselClassIDs": vessel_class_ids,
            "DateFrom": date_from.isoformat() if date_from else None,
        }

        response = get_single(
            connection=self.__connection,
            relative_url=query_url,
            cls=PortCongestionQueryResponse,
            query_string=params,
        )

        if not response:
            return None

        if response.query_errors is not None:
            raise RuntimeError(
                f"Port Congestion API errors occurred:"
                f"\n{[error for error in response.query_errors]}"
            )

        return response.time_series

__init__(connection=None)

Initializes PortCongestionAPI.

Parameters:

Name Type Description Default
connection Connection

API connection configuration. If not provided, the default connection method is used.

 None
Source code in signal_ocean/port_congestion/port_congestion_api.py 
19
20
21
22
23
24
25
26
27
def __init__(self, connection: Optional[Connection] = None):
    """Initializes PortCongestionAPI.

    Args:
        connection (Connection, optional): API connection configuration.
                                           If not provided, the default
                                           connection method is used.
    """
    self.__connection = connection or Connection()

query_port_congestion(ports=None, port_ids=None, level_0_areas=None, level_0_area_ids=None, level_1_areas=None, level_1_area_ids=None, countries=None, country_ids=None, vessel_types=None, vessel_type_ids=None, vessel_classes=None, vessel_class_ids=None, date_from=None)

Exposes Port Congestion's query endpoint.

Parameters:

Name Type Description Default
ports list

A list of Port Names. Defaults to None.

 None
port_ids list

A list of Port IDs. Defaults to None.

 None
level_0_areas list

A list of Level 0 Area names. Defaults to None.

 None
level_0_area_ids list

A list of Level 0 Area IDs. Defaults to None.

 None
level_1_areas list

A list of Level 1 Area names. Defaults to None.

 None
level_1_area_ids list

A list of Level 1 Area IDs. Defaults to None.

 None
countries list

A list of Country names. Defaults to None.

 None
country_ids list

A list of Country IDs. Defaults to None.

 None
vessel_types list

A list of Vessel Type names. Defaults to None.

 None
vessel_type_ids list

A list of Vessel Type IDs. Defaults to None.

 None
vessel_classes list

A list of Vessel Class names. Defaults to None.

 None
vessel_class_ids list

A list of Vessel Class IDs. Defaults to None.

 None
date_from date

A Date point to query from. Defaults to None.

 None

Raises:

Type Description
RuntimeError

In case of any request issue.

Returns:

Type Description
Optional[List[PortCongestionTimeSeriesEntry]]

List[PortCongestionTimeSeriesEntry], optional: A list of PortCongestionTimeSeriesEntry or None.
Source code in signal_ocean/port_congestion/port_congestion_api.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
def query_port_congestion(
    self,
    ports: Optional[List[str]] = None,
    port_ids: Optional[List[int]] = None,
    level_0_areas: Optional[List[str]] = None,
    level_0_area_ids: Optional[List[int]] = None,
    level_1_areas: Optional[List[str]] = None,
    level_1_area_ids: Optional[List[int]] = None,
    countries: Optional[List[str]] = None,
    country_ids: Optional[List[int]] = None,
    vessel_types: Optional[List[str]] = None,
    vessel_type_ids: Optional[List[int]] = None,
    vessel_classes: Optional[List[str]] = None,
    vessel_class_ids: Optional[List[int]] = None,
    date_from: Optional[date] = None,
) -> Optional[List[PortCongestionTimeSeriesEntry]]:
    """Exposes Port Congestion's `query` endpoint.

    Args:
        ports (list, optional): A list of Port Names. Defaults to None.
        port_ids (list, optional): A list of Port IDs. Defaults to None.
        level_0_areas (list, optional): A list of Level 0 Area names.
                                        Defaults to None.
        level_0_area_ids (list, optional): A list of Level 0 Area IDs.
                                           Defaults to None.
        level_1_areas (list, optional): A list of Level 1 Area names.
                                        Defaults to None.
        level_1_area_ids (list, optional): A list of Level 1 Area IDs.
                                           Defaults to None.
        countries (list, optional): A list of Country names.
                                    Defaults to None.
        country_ids (list, optional): A list of Country IDs.
                                      Defaults to None.
        vessel_types (list, optional): A list of Vessel Type names.
                                       Defaults to None.
        vessel_type_ids (list, optional): A list of Vessel Type IDs.
                                          Defaults to None.
        vessel_classes (list, optional): A list of Vessel Class names.
                                         Defaults to None.
        vessel_class_ids (list, optional): A list of Vessel Class IDs.
                                           Defaults to None.
        date_from (date, optional): A Date point to query from.
                                    Defaults to None.

    Raises:
        RuntimeError: In case of any request issue.

    Returns:
        List[PortCongestionTimeSeriesEntry], optional:
            A list of PortCongestionTimeSeriesEntry or None.
    """
    query_url = f"{self.base_url}/api/v1/query/"
    params = {
        "Ports": ports,
        "PortIDs": port_ids,
        "Level0Areas": level_0_areas,
        "Level0AreasIDs": level_0_area_ids,
        "Level1Areas": level_1_areas,
        "Level1AreaIDs": level_1_area_ids,
        "Countries": countries,
        "CountriesIDs": country_ids,
        "VesselTypes": vessel_types,
        "VesselTypeIDs": vessel_type_ids,
        "VesselClasses": vessel_classes,
        "VesselClassIDs": vessel_class_ids,
        "DateFrom": date_from.isoformat() if date_from else None,
    }

    response = get_single(
        connection=self.__connection,
        relative_url=query_url,
        cls=PortCongestionQueryResponse,
        query_string=params,
    )

    if not response:
        return None

    if response.query_errors is not None:
        raise RuntimeError(
            f"Port Congestion API errors occurred:"
            f"\n{[error for error in response.query_errors]}"
        )

    return response.time_series