Traffic API

Client for the HERE Traffic API.

TrafficApi

Bases: HEREApi

A python interface into the HERE Traffic API.

Source code in src/herepy/traffic_api.py

class TrafficApi(HEREApi):
    """A python interface into the HERE Traffic API."""

    def __init__(self, api_key: str = None, timeout: int = None):
        """
        Initialize the Traffic API client.

        Args:
          api_key (str):
            API key taken from HERE Developer Portal.
          timeout (int):
            Timeout limit for requests.

        """
        super().__init__(api_key, timeout)
        self._base_url = "https://traffic.ls.hereapi.com/traffic/6.1/"

    def __get(self, url, data):
        url = Utils.build_url(url, extra_params=data)
        response = requests.get(url, timeout=self._timeout)
        json_data = json.loads(response._content.decode("utf8"))
        if json_data.get("TRAFFIC_ITEMS") is not None:
            return TrafficIncidentResponse.new_from_jsondict(
                json_data,
                param_defaults={
                    "TIMESTAMP": None,
                    "VERSION": None,
                    "TRAFFIC_ITEMS": None,
                    "EXTENDED_COUNTRY_CODE": None,
                    "error": None,
                },
            )
        elif json_data.get("RWS") is not None:
            return TrafficFlowResponse.new_from_jsondict(
                json_data,
                param_defaults={
                    "RWS": None,
                    "CREATED_TIMESTAMP": None,
                    "VERSION": None,
                    "UNITS": None,
                },
            )
        elif json_data.get("Response") is not None:
            return TrafficFlowAvailabilityResponse.new_from_jsondict(json_data, param_defaults={"Response": None})
        else:
            error = self.__get_error_from_response(json_data)
            raise error

    def __get_error_from_response(self, json_data):
        if "error" in json_data:
            if json_data["error"] == "Unauthorized":
                return UnauthorizedError(json_data["error_description"])
        error_type = json_data.get("Type")
        error_message = json_data.get("Message", "Error occurred on " + sys._getframe(1).f_code.co_name)
        if error_type == "Invalid Request":
            return InvalidRequestError(error_message)
        else:
            return HEREError(error_message)

    def __prepare_str_values(self, enums: Iterable[Enum]):
        values = ""
        for enum in enums:
            values += enum.__str__() + ","
        values = values[:-1]
        return values

    def __prepare_criticality_int_values(self, criticality_enums: list[IncidentsCriticalityInt]):
        criticality_values = ""
        for criticality in criticality_enums:
            criticality_values += str(criticality.__int__()) + ","
        criticality_values = criticality_values[:-1]
        return criticality_values

    def __prepare_corridor_value(self, points: list[list[float]], width: int):
        corridor_str = ""
        for lat_long_pair in points:
            corridor_str += str.format("{0},{1};", lat_long_pair[0], lat_long_pair[1])
        corridor_str += str(width)
        return corridor_str

    def incidents_in_bounding_box(
        self,
        top_left: list[float],
        bottom_right: list[float],
        criticality: list[IncidentsCriticalityStr],
    ) -> TrafficIncidentResponse | None:
        """
        Request traffic incident information within specified area.

        Args:
          top_left (List):
            List contains latitude and longitude in order.
          bottom_right (List):
            List contains latitude and longitude in order.
          criticality (List):
            List of IncidentsCriticalityStr.

        Returns:
          TrafficIncidentResponse
        Raises:
          HEREError

        """
        data = {
            "bbox": str.format(
                "{0},{1};{2},{3}",
                top_left[0],
                top_left[1],
                bottom_right[0],
                bottom_right[1],
            ),
            "apiKey": self._api_key,
            "criticality": self.__prepare_str_values(enums=criticality),
        }
        return self.__get(self._base_url + "incidents.json", data)

    def incidents_in_corridor(self, points: list[list[float]], width: int) -> TrafficIncidentResponse | None:
        """
        Request traffic incidents for a defined route.

        Args:
          points (List):
            List contains lists of latitude and longitude pairs in order.
          width (int):
            Width of corridor.

        Returns:
          TrafficIncidentResponse
        Raises:
          HEREError

        """
        data = {
            "corridor": self.__prepare_corridor_value(points=points, width=width),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "incidents.json", data)

    def incidents_via_proximity(
        self,
        latitude: float,
        longitude: float,
        radius: int,
        criticality: list[IncidentsCriticalityInt],
    ) -> TrafficIncidentResponse | None:
        """
        Request traffic incident information within specified area.

        Args:
          latitude (float):
            Latitude of specified area.
          longitude (float):
            Longitude of specified area.
          radius (int):
            Radius of area in meters.
          criticality (List):
            List of IncidentsCriticalityInt.

        Returns:
          TrafficIncidentResponse
        Raises:
          HEREError

        """
        data = {
            "prox": str.format("{0},{1},{2}", latitude, longitude, radius),
            "criticality": self.__prepare_criticality_int_values(criticality_enums=criticality),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "incidents.json", data)

    def flow_using_quadkey(self, quadkey: str) -> TrafficFlowResponse | None:
        """
        Request traffic flow information using a quadkey.

        Args:
          quadkey (str):
            The quadkey unique defines a region of the globe using a standard addressing algortihm.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "quadkey": quadkey,
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_within_boundingbox(
        self,
        top_left: list[float],
        bottom_right: list[float],
        response_attributes: str = "sh,fc",
    ) -> TrafficFlowResponse | None:
        """
        Request traffic flow information within specified area.

        Args:
          top_left (List):
            List contains latitude and longitude in order.
          bottom_right (List):
            List contains latitude and longitude in order.
          response_attributes (String):
            String containing response attributes separated by commas.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "bbox": str.format(
                "{0},{1};{2},{3}",
                top_left[0],
                top_left[1],
                bottom_right[0],
                bottom_right[1],
            ),
            "apiKey": self._api_key,
            "responseattributes": response_attributes,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_using_proximity(self, latitude: float, longitude: float, distance: int) -> TrafficFlowResponse | None:
        """
        Request traffic flow for a circle around a defined point.

        Args:
          latitude (float):
            List contains latitude and longitude in order.
          longitude (float):
            List contains latitude and longitude in order.
          distance (int):
            Extending a distance of metres in all directions.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "prox": str.format(
                "{0},{1},{2}",
                latitude,
                longitude,
                distance,
            ),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_using_proximity_returning_additional_attributes(
        self,
        latitude: float,
        longitude: float,
        distance: int,
        attributes: list[FlowProximityAdditionalAttributes],
    ) -> TrafficFlowResponse | None:
        """
        Request traffic flow information using proximity, returning shape and functional class.

        Args:
          latitude (float):
            List contains latitude and longitude in order.
          longitude (float):
            List contains latitude and longitude in order.
          distance (int):
            Extending a distance of metres in all directions.
          attributes (List):
            List that contains FlowProximityAdditionalAttributes.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "prox": str.format(
                "{0},{1},{2}",
                latitude,
                longitude,
                distance,
            ),
            "responseattibutes": self.__prepare_str_values(enums=attributes),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_with_minimum_jam_factor(
        self, top_left: list[float], bottom_right: list[float], min_jam_factor: int = 7
    ) -> TrafficFlowResponse | None:
        """
        Request traffic flow information in specified area with a jam factor.

        Args:
          top_left (List):
            List contains latitude and longitude in order.
          bottom_right (List):
            List contains latitude and longitude in order.
          min_jam_factor (int):
            Severe congestion with a jam factor greater than 7.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "bbox": str.format(
                "{0},{1};{2},{3}",
                top_left[0],
                top_left[1],
                bottom_right[0],
                bottom_right[1],
            ),
            "minjamfactor": str.format("{0}", min_jam_factor),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_in_corridor(self, points: list[list[float]], width: int) -> TrafficFlowResponse | None:
        """
        Request traffic flow for a defined route.

        Args:
          points (List):
            List contains lists of latitude and longitude pairs in order.
          width (int):
            Width of corridor (in meters).

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "corridor": self.__prepare_corridor_value(points=points, width=width),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

    def flow_availability_data(self) -> TrafficFlowAvailabilityResponse | None:
        """
        Flow availability requests allow you to see what traffic flow coverage exists in the current Traffic API.

        Returns:
          TrafficFlowAvailabilityResponse
        Raises:
          HEREError

        """
        data = {
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flowavailability.json", data)

    def additional_attributes(self, quadkey: str, attributes: list[FlowProximityAdditionalAttributes]) -> list[TrafficFlowResponse]:
        """
        Request traffic flow including shape and functional class information.

        Args:
          quadkey (str):
            The quadkey unique defines a region of the globe using a standard addressing algortihm.
          attributes (List):
            List that contains FlowProximityAdditionalAttributes.

        Returns:
          TrafficFlowResponse
        Raises:
          HEREError

        """
        data = {
            "quadkey": quadkey,
            "responseattibutes": self.__prepare_str_values(enums=attributes),
            "apiKey": self._api_key,
        }
        return self.__get(self._base_url + "flow.json", data)

__init__(api_key=None, timeout=None)

Initialize the Traffic API client.

Parameters:

Name	Type	Description	Default
api_key	str	API key taken from HERE Developer Portal.	None
timeout	int	Timeout limit for requests.	None

Source code in src/herepy/traffic_api.py

def __init__(self, api_key: str = None, timeout: int = None):
    """
    Initialize the Traffic API client.

    Args:
      api_key (str):
        API key taken from HERE Developer Portal.
      timeout (int):
        Timeout limit for requests.

    """
    super().__init__(api_key, timeout)
    self._base_url = "https://traffic.ls.hereapi.com/traffic/6.1/"

additional_attributes(quadkey, attributes)

Request traffic flow including shape and functional class information.

Parameters:

Name	Type	Description	Default
quadkey	str	The quadkey unique defines a region of the globe using a standard addressing algortihm.	required
attributes	List	List that contains FlowProximityAdditionalAttributes.	required

Returns:

Type	Description
list[TrafficFlowResponse]	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def additional_attributes(self, quadkey: str, attributes: list[FlowProximityAdditionalAttributes]) -> list[TrafficFlowResponse]:
    """
    Request traffic flow including shape and functional class information.

    Args:
      quadkey (str):
        The quadkey unique defines a region of the globe using a standard addressing algortihm.
      attributes (List):
        List that contains FlowProximityAdditionalAttributes.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "quadkey": quadkey,
        "responseattibutes": self.__prepare_str_values(enums=attributes),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_availability_data()

Flow availability requests allow you to see what traffic flow coverage exists in the current Traffic API.

Returns:

Type	Description
TrafficFlowAvailabilityResponse | None	TrafficFlowAvailabilityResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_availability_data(self) -> TrafficFlowAvailabilityResponse | None:
    """
    Flow availability requests allow you to see what traffic flow coverage exists in the current Traffic API.

    Returns:
      TrafficFlowAvailabilityResponse
    Raises:
      HEREError

    """
    data = {
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flowavailability.json", data)

flow_in_corridor(points, width)

Request traffic flow for a defined route.

Parameters:

Name	Type	Description	Default
points	List	List contains lists of latitude and longitude pairs in order.	required
width	int	Width of corridor (in meters).	required

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_in_corridor(self, points: list[list[float]], width: int) -> TrafficFlowResponse | None:
    """
    Request traffic flow for a defined route.

    Args:
      points (List):
        List contains lists of latitude and longitude pairs in order.
      width (int):
        Width of corridor (in meters).

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "corridor": self.__prepare_corridor_value(points=points, width=width),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_using_proximity(latitude, longitude, distance)

Request traffic flow for a circle around a defined point.

Parameters:

Name	Type	Description	Default
latitude	float	List contains latitude and longitude in order.	required
longitude	float	List contains latitude and longitude in order.	required
distance	int	Extending a distance of metres in all directions.	required

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_using_proximity(self, latitude: float, longitude: float, distance: int) -> TrafficFlowResponse | None:
    """
    Request traffic flow for a circle around a defined point.

    Args:
      latitude (float):
        List contains latitude and longitude in order.
      longitude (float):
        List contains latitude and longitude in order.
      distance (int):
        Extending a distance of metres in all directions.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "prox": str.format(
            "{0},{1},{2}",
            latitude,
            longitude,
            distance,
        ),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_using_proximity_returning_additional_attributes(latitude, longitude, distance, attributes)

Request traffic flow information using proximity, returning shape and functional class.

Parameters:

Name	Type	Description	Default
latitude	float	List contains latitude and longitude in order.	required
longitude	float	List contains latitude and longitude in order.	required
distance	int	Extending a distance of metres in all directions.	required
attributes	List	List that contains FlowProximityAdditionalAttributes.	required

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_using_proximity_returning_additional_attributes(
    self,
    latitude: float,
    longitude: float,
    distance: int,
    attributes: list[FlowProximityAdditionalAttributes],
) -> TrafficFlowResponse | None:
    """
    Request traffic flow information using proximity, returning shape and functional class.

    Args:
      latitude (float):
        List contains latitude and longitude in order.
      longitude (float):
        List contains latitude and longitude in order.
      distance (int):
        Extending a distance of metres in all directions.
      attributes (List):
        List that contains FlowProximityAdditionalAttributes.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "prox": str.format(
            "{0},{1},{2}",
            latitude,
            longitude,
            distance,
        ),
        "responseattibutes": self.__prepare_str_values(enums=attributes),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_using_quadkey(quadkey)

Request traffic flow information using a quadkey.

Parameters:

Name	Type	Description	Default
quadkey	str	The quadkey unique defines a region of the globe using a standard addressing algortihm.	required

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_using_quadkey(self, quadkey: str) -> TrafficFlowResponse | None:
    """
    Request traffic flow information using a quadkey.

    Args:
      quadkey (str):
        The quadkey unique defines a region of the globe using a standard addressing algortihm.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "quadkey": quadkey,
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_with_minimum_jam_factor(top_left, bottom_right, min_jam_factor=7)

Request traffic flow information in specified area with a jam factor.

Parameters:

Name	Type	Description	Default
top_left	List	List contains latitude and longitude in order.	required
bottom_right	List	List contains latitude and longitude in order.	required
min_jam_factor	int	Severe congestion with a jam factor greater than 7.	7

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_with_minimum_jam_factor(
    self, top_left: list[float], bottom_right: list[float], min_jam_factor: int = 7
) -> TrafficFlowResponse | None:
    """
    Request traffic flow information in specified area with a jam factor.

    Args:
      top_left (List):
        List contains latitude and longitude in order.
      bottom_right (List):
        List contains latitude and longitude in order.
      min_jam_factor (int):
        Severe congestion with a jam factor greater than 7.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "bbox": str.format(
            "{0},{1};{2},{3}",
            top_left[0],
            top_left[1],
            bottom_right[0],
            bottom_right[1],
        ),
        "minjamfactor": str.format("{0}", min_jam_factor),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "flow.json", data)

flow_within_boundingbox(top_left, bottom_right, response_attributes='sh,fc')

Request traffic flow information within specified area.

Parameters:

Name	Type	Description	Default
top_left	List	List contains latitude and longitude in order.	required
bottom_right	List	List contains latitude and longitude in order.	required
response_attributes	String	String containing response attributes separated by commas.	'sh,fc'

Returns:

Type	Description
TrafficFlowResponse | None	TrafficFlowResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def flow_within_boundingbox(
    self,
    top_left: list[float],
    bottom_right: list[float],
    response_attributes: str = "sh,fc",
) -> TrafficFlowResponse | None:
    """
    Request traffic flow information within specified area.

    Args:
      top_left (List):
        List contains latitude and longitude in order.
      bottom_right (List):
        List contains latitude and longitude in order.
      response_attributes (String):
        String containing response attributes separated by commas.

    Returns:
      TrafficFlowResponse
    Raises:
      HEREError

    """
    data = {
        "bbox": str.format(
            "{0},{1};{2},{3}",
            top_left[0],
            top_left[1],
            bottom_right[0],
            bottom_right[1],
        ),
        "apiKey": self._api_key,
        "responseattributes": response_attributes,
    }
    return self.__get(self._base_url + "flow.json", data)

incidents_in_bounding_box(top_left, bottom_right, criticality)

Request traffic incident information within specified area.

Parameters:

Name	Type	Description	Default
top_left	List	List contains latitude and longitude in order.	required
bottom_right	List	List contains latitude and longitude in order.	required
criticality	List	List of IncidentsCriticalityStr.	required

Returns:

Type	Description
TrafficIncidentResponse | None	TrafficIncidentResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def incidents_in_bounding_box(
    self,
    top_left: list[float],
    bottom_right: list[float],
    criticality: list[IncidentsCriticalityStr],
) -> TrafficIncidentResponse | None:
    """
    Request traffic incident information within specified area.

    Args:
      top_left (List):
        List contains latitude and longitude in order.
      bottom_right (List):
        List contains latitude and longitude in order.
      criticality (List):
        List of IncidentsCriticalityStr.

    Returns:
      TrafficIncidentResponse
    Raises:
      HEREError

    """
    data = {
        "bbox": str.format(
            "{0},{1};{2},{3}",
            top_left[0],
            top_left[1],
            bottom_right[0],
            bottom_right[1],
        ),
        "apiKey": self._api_key,
        "criticality": self.__prepare_str_values(enums=criticality),
    }
    return self.__get(self._base_url + "incidents.json", data)

incidents_in_corridor(points, width)

Request traffic incidents for a defined route.

Parameters:

Name	Type	Description	Default
points	List	List contains lists of latitude and longitude pairs in order.	required
width	int	Width of corridor.	required

Returns:

Type	Description
TrafficIncidentResponse | None	TrafficIncidentResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def incidents_in_corridor(self, points: list[list[float]], width: int) -> TrafficIncidentResponse | None:
    """
    Request traffic incidents for a defined route.

    Args:
      points (List):
        List contains lists of latitude and longitude pairs in order.
      width (int):
        Width of corridor.

    Returns:
      TrafficIncidentResponse
    Raises:
      HEREError

    """
    data = {
        "corridor": self.__prepare_corridor_value(points=points, width=width),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "incidents.json", data)

incidents_via_proximity(latitude, longitude, radius, criticality)

Request traffic incident information within specified area.

Parameters:

Name	Type	Description	Default
latitude	float	Latitude of specified area.	required
longitude	float	Longitude of specified area.	required
radius	int	Radius of area in meters.	required
criticality	List	List of IncidentsCriticalityInt.	required

Returns:

Type	Description
TrafficIncidentResponse | None	TrafficIncidentResponse

Raises: HEREError

Source code in src/herepy/traffic_api.py

def incidents_via_proximity(
    self,
    latitude: float,
    longitude: float,
    radius: int,
    criticality: list[IncidentsCriticalityInt],
) -> TrafficIncidentResponse | None:
    """
    Request traffic incident information within specified area.

    Args:
      latitude (float):
        Latitude of specified area.
      longitude (float):
        Longitude of specified area.
      radius (int):
        Radius of area in meters.
      criticality (List):
        List of IncidentsCriticalityInt.

    Returns:
      TrafficIncidentResponse
    Raises:
      HEREError

    """
    data = {
        "prox": str.format("{0},{1},{2}", latitude, longitude, radius),
        "criticality": self.__prepare_criticality_int_values(criticality_enums=criticality),
        "apiKey": self._api_key,
    }
    return self.__get(self._base_url + "incidents.json", data)