Skip to content

Package Documentation

Geos API Package.

Classes:

Name Description
GeosAPI

Represents Signal's Geos API.

GeosAPI

Represents Signal's Geos API.

Source code in signal_ocean/geos/geos_api.py 
 13
 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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class GeosAPI:
    """Represents Signal's Geos API."""

    relative_url = "geos-api/v2/"
    default_pit = str(date.today())

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

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

    def get_areas(self, areaId: Optional[int] = None) -> Tuple[Area, ...]:
        """Retrieves all available areas.

        Args:
            areaId: area identifier to filter and return only a
                specific area.

        Returns:
            A tuple of all available areas.
        """
        endpoint = (
            "areas/all" if areaId is None else f"areas/{areaId}"
        )

        url = urljoin(GeosAPI.relative_url, endpoint)
        response = get_single(self.__connection, url, AreasPagedResponse)

        if response is not None and response.data is not None:
            return response.data
        else:
            return tuple([])

    def get_geoAssets(
        self,
        geoAssetId: Optional[int] = None
    ) -> Tuple[GeoAsset, ...]:
        """Retrieves all available geo assets.

        Args:
            geoAssetId: geo asset identifier to filter and return only a
                specific geo asset.

        Returns:
            A tuple of all available geo asset.
        """
        endpoint = (
            "geoAssets/all" if geoAssetId is None
            else f"geoAssets/{geoAssetId}"
        )
        rename_keys = {
            "AreaIDLevel0": "area_id_level0",
            "AreaIDLevel1": "area_id_level1",
            "AreaIDLevel2": "area_id_level2",
            "AreaIDLevel3": "area_id_level3",
            "CountryCodeISO3": "country_code_iso3"}
        url = urljoin(GeosAPI.relative_url, endpoint)
        response = get_single(
            self.__connection,
            url,
            GeoAssetsPagedResponse,
            rename_keys=rename_keys)
        if response is not None and response.data is not None:
            return response.data
        else:
            return tuple([])

    def get_countries(
        self,
        countryId: Optional[int] = None
    ) -> Tuple[Country, ...]:
        """Retrieves all available countries.

        Args:
            countryId: country identifier to filter and return only a
                specific country.

        Returns:
            A tuple of all available countries.
        """
        endpoint = (
            "countries/all" if countryId is None else f"countries/{countryId}"
        )
        url = urljoin(GeosAPI.relative_url, endpoint)
        response = get_single(self.__connection, url, CountriesPagedResponse)
        if response is not None and response.data is not None:
            return response.data
        else:
            return tuple([])

    def get_ports(self, portId: Optional[int] = None) -> Tuple[Port, ...]:
        """Retrieves all available ports.

        Args:
            portId: port identifier to filter and return only a
                specific port.

        Returns:
            A tuple of all available ports.
        """
        endpoint = (
            "ports/all" if portId is None else f"ports/{portId}"
        )
        rename_keys = {
            "AreaIDLevel0": "area_id_level0",
            "AreaIDLevel1": "area_id_level1",
            "AreaIDLevel2": "area_id_level2",
            "AreaIDLevel3": "area_id_level3"}
        url = urljoin(GeosAPI.relative_url, endpoint)
        response = get_single(
            self.__connection,
            url,
            PortsPagedResponse,
            rename_keys=rename_keys)
        if response is not None and response.data is not None:
            return response.data
        else:
            return tuple([])

__init__(connection=None)

Initializes GeosAPI.

Parameters:

Name Type Description Default
connection Optional[Connection]

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

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

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

get_areas(areaId=None)

Retrieves all available areas.

Parameters:

Name Type Description Default
areaId Optional[int]

area identifier to filter and return only a specific area.

 None

Returns:

Type Description
Tuple[Area, ...]

A tuple of all available areas.
Source code in signal_ocean/geos/geos_api.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def get_areas(self, areaId: Optional[int] = None) -> Tuple[Area, ...]:
    """Retrieves all available areas.

    Args:
        areaId: area identifier to filter and return only a
            specific area.

    Returns:
        A tuple of all available areas.
    """
    endpoint = (
        "areas/all" if areaId is None else f"areas/{areaId}"
    )

    url = urljoin(GeosAPI.relative_url, endpoint)
    response = get_single(self.__connection, url, AreasPagedResponse)

    if response is not None and response.data is not None:
        return response.data
    else:
        return tuple([])

get_countries(countryId=None)

Retrieves all available countries.

Parameters:

Name Type Description Default
countryId Optional[int]

country identifier to filter and return only a specific country.

 None

Returns:

Type Description
Tuple[Country, ...]

A tuple of all available countries.
Source code in signal_ocean/geos/geos_api.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_countries(
    self,
    countryId: Optional[int] = None
) -> Tuple[Country, ...]:
    """Retrieves all available countries.

    Args:
        countryId: country identifier to filter and return only a
            specific country.

    Returns:
        A tuple of all available countries.
    """
    endpoint = (
        "countries/all" if countryId is None else f"countries/{countryId}"
    )
    url = urljoin(GeosAPI.relative_url, endpoint)
    response = get_single(self.__connection, url, CountriesPagedResponse)
    if response is not None and response.data is not None:
        return response.data
    else:
        return tuple([])

get_geoAssets(geoAssetId=None)

Retrieves all available geo assets.

Parameters:

Name Type Description Default
geoAssetId Optional[int]

geo asset identifier to filter and return only a specific geo asset.

 None

Returns:

Type Description
Tuple[GeoAsset, ...]

A tuple of all available geo asset.
Source code in signal_ocean/geos/geos_api.py 
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
def get_geoAssets(
    self,
    geoAssetId: Optional[int] = None
) -> Tuple[GeoAsset, ...]:
    """Retrieves all available geo assets.

    Args:
        geoAssetId: geo asset identifier to filter and return only a
            specific geo asset.

    Returns:
        A tuple of all available geo asset.
    """
    endpoint = (
        "geoAssets/all" if geoAssetId is None
        else f"geoAssets/{geoAssetId}"
    )
    rename_keys = {
        "AreaIDLevel0": "area_id_level0",
        "AreaIDLevel1": "area_id_level1",
        "AreaIDLevel2": "area_id_level2",
        "AreaIDLevel3": "area_id_level3",
        "CountryCodeISO3": "country_code_iso3"}
    url = urljoin(GeosAPI.relative_url, endpoint)
    response = get_single(
        self.__connection,
        url,
        GeoAssetsPagedResponse,
        rename_keys=rename_keys)
    if response is not None and response.data is not None:
        return response.data
    else:
        return tuple([])

get_ports(portId=None)

Retrieves all available ports.

Parameters:

Name Type Description Default
portId Optional[int]

port identifier to filter and return only a specific port.

 None

Returns:

Type Description
Tuple[Port, ...]

A tuple of all available ports.
Source code in signal_ocean/geos/geos_api.py 
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
def get_ports(self, portId: Optional[int] = None) -> Tuple[Port, ...]:
    """Retrieves all available ports.

    Args:
        portId: port identifier to filter and return only a
            specific port.

    Returns:
        A tuple of all available ports.
    """
    endpoint = (
        "ports/all" if portId is None else f"ports/{portId}"
    )
    rename_keys = {
        "AreaIDLevel0": "area_id_level0",
        "AreaIDLevel1": "area_id_level1",
        "AreaIDLevel2": "area_id_level2",
        "AreaIDLevel3": "area_id_level3"}
    url = urljoin(GeosAPI.relative_url, endpoint)
    response = get_single(
        self.__connection,
        url,
        PortsPagedResponse,
        rename_keys=rename_keys)
    if response is not None and response.data is not None:
        return response.data
    else:
        return tuple([])