implement traffic density endpoint

Author: Guillaume Latour

pyproject.toml

@@ -22,7 +22,7 @@ maintainers = [
 name = "tomtom_api"
 readme = "readme.md"
 requires-python = ">=3.8"
-version = "2024.6.27"
+version = "2024.7.4"
 
 [project.urls]
 repository = "https://stash.macq.eu/projects/RDDS/repos/tomtom-api"

src/tomtom_api/client.py

@@ -11,6 +11,7 @@
 from tomtom_api.traffic_stats.models.geospatial import TomtomNetwork, TomtomRoad
 from tomtom_api.traffic_stats.models.jobs.area import TomtomAreaJob
 from tomtom_api.traffic_stats.models.jobs.route import TomtomRouteJob
+from tomtom_api.traffic_stats.models.jobs.traffic_density import TomtomTrafficDensityJob
 from tomtom_api.traffic_stats.models.responses import (
     JsonIpResponse,
     TomtomErrorResponse,
@@ -147,7 +148,7 @@ def route_analysis(
         date_ranges: List[TomtomDateRange],
         time_sets: List[TomtomTimeSet],
         accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO",
-        map_version: Optional[float] = 2023.03,
+        map_version: Optional[float] = None,
         average_sample_size_threshold: Optional[int] = None,
     ) -> TomtomResponseAnalysis:
         """
@@ -176,7 +177,7 @@ def area_analysis(
         date_range: TomtomDateRange,
         time_sets: List[TomtomTimeSet],
         accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO",
-        map_version: Optional[float] = 2023.03,
+        map_version: Optional[float] = None,
         average_sample_size_threshold: Optional[int] = None,
     ) -> TomtomResponseAnalysis:
         job = TomtomAreaJob(
@@ -191,10 +192,33 @@ def area_analysis(
         )
         return self.post_job_area_analysis(job)
 
+    def traffic_density(
+        self,
+        job_name: str,
+        distance_unit: Literal["KILOMETERS", "MILES"],
+        network: TomtomNetwork,
+        date_range: TomtomDateRange,
+        time_sets: List[TomtomTimeSet],
+        accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO",
+        map_version: Optional[float] = None,
+        average_sample_size_threshold: Optional[int] = None,
+    ) -> TomtomResponseAnalysis:
+        job = TomtomTrafficDensityJob(
+            job_name=job_name,
+            distance_unit=distance_unit,
+            network=network,
+            date_range=date_range,
+            time_sets=time_sets,
+            accept_mode=accept_mode,
+            map_version=map_version,
+            average_sample_size_threshold=average_sample_size_threshold,
+        )
+        return self.post_job_traffic_density(job)
+
     def find_route(
         self,
         road: TomtomRoad,
-        map_version: Optional[str] = "2016.12",
+        map_version: Optional[str] = None,
         map_type: Optional[str] = "DSEG_NOSPLIT",
     ) -> TomtomResponseRouteFound:
         """
@@ -234,6 +258,15 @@ def post_job_area_analysis(self, job: TomtomAreaJob) -> TomtomResponseAnalysis:
         tomtom_response = TomtomResponseAnalysis.from_dict(request_response.json())
         return tomtom_response
 
+    def post_job_traffic_density(
+        self, job: TomtomTrafficDensityJob
+    ) -> TomtomResponseAnalysis:
+        url = f"https://{self.base_url}/traffic/trafficstats/trafficdensity/{self.version}?key={self.key}"
+
+        request_response = self.request("post", url, data=json.dumps(job.to_dict()))
+        tomtom_response = TomtomResponseAnalysis.from_dict(request_response.json())
+        return tomtom_response
+
     def status(self, job_id: int) -> TomtomResponseStatus:
         """When a job has been initiated via the API request it is possible to check the status.
 

src/tomtom_api/traffic_stats/models/jobs/area.py

@@ -18,7 +18,7 @@ def __init__(
         time_sets: List[TomtomTimeSet],
         distance_unit: Literal['KILOMETERS', 'MILES'] = 'KILOMETERS',
         accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO',
-        map_version: Optional[float] = 2020.09,
+        map_version: Optional[float] = None,
         average_sample_size_threshold: Optional[int] = None
     ):
         self.network = network

src/tomtom_api/traffic_stats/models/jobs/base.py

@@ -9,24 +9,25 @@
 
 class TomtomJob:
     """Data structure allowing the manipulation of the tomtom jobs"""
+
     job_name: str
     date_ranges: List[TomtomDateRange]
     time_sets: List[TomtomTimeSet]
-    distance_unit: str = 'KILOMETERS'
-    accept_mode: Optional[str] = 'AUTO'
+    distance_unit: str = "KILOMETERS"
+    accept_mode: Optional[str] = "AUTO"
     # see here for more info about map versions
     # https://developer.tomtom.com/traffic-stats/documentation/api/available-maps
-    map_version: Optional[float] = 2020.09
+    map_version: Optional[float] = None
     average_sample_size_threshold: Optional[int] = None
 
     def __init__(
         self,
         job_name: str,
         time_sets: List[TomtomTimeSet],
-        distance_unit: Literal['KILOMETERS', 'MILES'] = 'KILOMETERS',
-        accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO',
-        map_version: Optional[float] = 2020.09,
-        average_sample_size_threshold: Optional[int] = None
+        distance_unit: Literal["KILOMETERS", "MILES"] = "KILOMETERS",
+        accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO",
+        map_version: Optional[float] = None,
+        average_sample_size_threshold: Optional[int] = None,
     ):
         """Data structure containing all the job information.
 
@@ -59,16 +60,16 @@ def __init__(
 
         if len(time_sets) > MAX_TIME_SETS_COUNT:
             raise ValueError(
-                f'Impossible to query for more than {MAX_TIME_SETS_COUNT} time sets, ({len(time_sets)} given)'
+                f"Impossible to query for more than {MAX_TIME_SETS_COUNT} time sets, ({len(time_sets)} given)"
             )
         self.time_sets = time_sets
 
-        if distance_unit.upper() not in ['KILOMETERS', 'MILES']:
-            raise ValueError(f'The given `distance_unit` is invalid ({distance_unit})')
+        if distance_unit.upper() not in ["KILOMETERS", "MILES"]:
+            raise ValueError(f"The given `distance_unit` is invalid ({distance_unit})")
         self.distance_unit = str(distance_unit).upper()
 
-        if accept_mode is not None and accept_mode.upper() not in ['AUTO', 'MANUAL']:
-            raise ValueError(f'The given `accept_mode` is invalid ({accept_mode})')
+        if accept_mode is not None and accept_mode.upper() not in ["AUTO", "MANUAL"]:
+            raise ValueError(f"The given `accept_mode` is invalid ({accept_mode})")
         self.accept_mode = accept_mode
 
         self.map_version = map_version
@@ -77,19 +78,23 @@ def __init__(
     @classmethod
     def from_dict(cls, dict_object: Dict[str, Any]) -> TomtomJob:
         return cls(
-            job_name=dict_object['jobName'],
-            time_sets=[TomtomTimeSet.from_dict(t) for t in dict_object['timeSets']],
-            distance_unit=dict_object['distanceUnit'],
-            accept_mode=dict_object['acceptMode'],
-            map_version=dict_object['mapVersion'],
-            average_sample_size_threshold=None if 'averageSampleSizeThreshold' not in dict_object else dict_object[
-                'averageSampleSizeThreshold']
+            job_name=dict_object["jobName"],
+            time_sets=[TomtomTimeSet.from_dict(t) for t in dict_object["timeSets"]],
+            distance_unit=dict_object["distanceUnit"],
+            accept_mode=dict_object["acceptMode"],
+            map_version=dict_object["mapVersion"],
+            average_sample_size_threshold=(
+                None
+                if "averageSampleSizeThreshold" not in dict_object
+                else dict_object["averageSampleSizeThreshold"]
+            ),
         )
 
-    def md5(self, salt: str = '') -> str:
+    def md5(self, salt: str = "") -> str:
         import hashlib
+
         json = self.to_json()
-        md5_hash = hashlib.md5(f'{json}{salt}'.encode('utf-8'))
+        md5_hash = hashlib.md5(f"{json}{salt}".encode("utf-8"))
         return md5_hash.hexdigest()
 
     def to_dict(self) -> Dict[str, Any]:
@@ -102,21 +107,21 @@ def to_dict(self) -> Dict[str, Any]:
             The dictionary with the keys specified in the documentation.
         """
         job_dict = {
-            'jobName': self.job_name,
-            'distanceUnit': self.distance_unit,
-            'mapVersion': None,
-            'acceptMode': None,
-            'timeSets': [t.to_dict() for t in self.time_sets],
-            'averageSampleSizeThreshold': None
+            "jobName": self.job_name,
+            "distanceUnit": self.distance_unit,
+            "mapVersion": None,
+            "acceptMode": None,
+            "timeSets": [t.to_dict() for t in self.time_sets],
+            "averageSampleSizeThreshold": None,
         }
 
         # manage optional fields
         if self.accept_mode is not None:
-            job_dict['acceptMode'] = self.accept_mode
+            job_dict["acceptMode"] = self.accept_mode
         if self.map_version is not None:
-            job_dict['mapVersion'] = self.map_version
+            job_dict["mapVersion"] = self.map_version
         if self.average_sample_size_threshold is not None:
-            job_dict['averageSampleSizeThreshold'] = self.average_sample_size_threshold
+            job_dict["averageSampleSizeThreshold"] = self.average_sample_size_threshold
 
         job_dict = {k: v for k, v in job_dict.items() if v is not None}
 

src/tomtom_api/traffic_stats/models/jobs/route.py

@@ -2,7 +2,7 @@
 
 from typing import Any, Dict, List, Literal, Optional
 
-from tomtom_api.traffic_stats import (MAX_DATE_RANGE_COUNT, MAX_ROAD_COUNT)
+from tomtom_api.traffic_stats import MAX_DATE_RANGE_COUNT, MAX_ROAD_COUNT
 from tomtom_api.traffic_stats.models.geospatial import TomtomRoad
 from tomtom_api.traffic_stats.models.jobs.base import TomtomJob
 from tomtom_api.traffic_stats.models.time import TomtomDateRange, TomtomTimeSet
@@ -17,10 +17,10 @@ def __init__(
         routes: List[TomtomRoad],
         date_ranges: List[TomtomDateRange],
         time_sets: List[TomtomTimeSet],
-        distance_unit: Literal['KILOMETERS', 'MILES'] = 'KILOMETERS',
-        accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO',
-        map_version: Optional[float] = 2020.09,
-        average_sample_size_threshold: Optional[int] = None
+        distance_unit: Literal["KILOMETERS", "MILES"] = "KILOMETERS",
+        accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO",
+        map_version: Optional[float] = None,
+        average_sample_size_threshold: Optional[int] = None,
     ):
         """Data structure containing all the job information.
 
@@ -55,12 +55,14 @@ def __init__(
         """
 
         if len(routes) > MAX_ROAD_COUNT:
-            raise ValueError(f'Impossible to query for more than {MAX_ROAD_COUNT} roads, ({len(routes)} given)')
+            raise ValueError(
+                f"Impossible to query for more than {MAX_ROAD_COUNT} roads, ({len(routes)} given)"
+            )
         self.routes = routes
 
         if len(date_ranges) > MAX_DATE_RANGE_COUNT:
             raise ValueError(
-                f'Impossible to query for more than {MAX_DATE_RANGE_COUNT} date range, ({len(date_ranges)} given)'
+                f"Impossible to query for more than {MAX_DATE_RANGE_COUNT} date range, ({len(date_ranges)} given)"
             )
         self.date_ranges = date_ranges
 
@@ -70,21 +72,26 @@ def __init__(
             distance_unit=distance_unit,
             accept_mode=accept_mode,
             map_version=map_version,
-            average_sample_size_threshold=average_sample_size_threshold
+            average_sample_size_threshold=average_sample_size_threshold,
         )
 
     @classmethod
     def from_dict(cls, dict_object: Dict[str, Any]) -> TomtomRouteJob:
         return cls(
-            job_name=dict_object['jobName'],
-            routes=[TomtomRoad.from_dict(r) for r in dict_object['routes']],
-            date_ranges=[TomtomDateRange.from_dict(d) for d in dict_object['dateRanges']],
-            time_sets=[TomtomTimeSet.from_dict(t) for t in dict_object['timeSets']],
-            distance_unit=dict_object['distanceUnit'],
-            accept_mode=dict_object['acceptMode'],
-            map_version=dict_object['mapVersion'],
-            average_sample_size_threshold=None if 'averageSampleSizeThreshold' not in dict_object else dict_object[
-                'averageSampleSizeThreshold']
+            job_name=dict_object["jobName"],
+            routes=[TomtomRoad.from_dict(r) for r in dict_object["routes"]],
+            date_ranges=[
+                TomtomDateRange.from_dict(d) for d in dict_object["dateRanges"]
+            ],
+            time_sets=[TomtomTimeSet.from_dict(t) for t in dict_object["timeSets"]],
+            distance_unit=dict_object["distanceUnit"],
+            accept_mode=dict_object["acceptMode"],
+            map_version=dict_object["mapVersion"],
+            average_sample_size_threshold=(
+                None
+                if "averageSampleSizeThreshold" not in dict_object
+                else dict_object["averageSampleSizeThreshold"]
+            ),
         )
 
     def to_dict(self) -> Dict[str, Any]:
@@ -98,8 +105,8 @@ def to_dict(self) -> Dict[str, Any]:
         """
         job_dict = {
             **super().to_dict(),
-            'dateRanges': [d.to_dict() for d in self.date_ranges],
-            'routes': [r.to_dict() for r in self.routes]
+            "dateRanges": [d.to_dict() for d in self.date_ranges],
+            "routes": [r.to_dict() for r in self.routes],
         }
 
         return job_dict

src/tomtom_api/traffic_stats/models/jobs/traffic_density.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+from tomtom_api.traffic_stats.models.geospatial import TomtomNetwork
+from tomtom_api.traffic_stats.models.jobs.base import TomtomJob
+from tomtom_api.traffic_stats.models.time import TomtomDateRange, TomtomTimeSet
+
+
+class TomtomTrafficDensityJob(TomtomJob):
+    network: TomtomNetwork
+
+    def __init__(
+        self,
+        job_name: str,
+        network: TomtomNetwork,
+        date_range: TomtomDateRange,
+        time_sets: List[TomtomTimeSet],
+        distance_unit: Literal['KILOMETERS', 'MILES'] = 'KILOMETERS',
+        accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO',
+        map_version: Optional[float] = None,
+        average_sample_size_threshold: Optional[int] = None
+    ):
+        self.network = network
+        self.date_range = date_range
+        super().__init__(job_name, time_sets, distance_unit, accept_mode, map_version, average_sample_size_threshold)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Transform this data type to a python dictionary.
+        This dict is meant to be used to generate the payload given to some Tomtom API endpoints.
+
+        Returns
+        -------
+        Dict[str, float]
+            The dictionary with the keys specified in the documentation.
+        """
+        job_dict = {
+            **super().to_dict(),
+            'dateRange': self.date_range.to_dict(),
+            'network': self.network.to_dict()
+        }
+
+        return job_dict
+
+    @classmethod
+    def from_dict(cls, dict_object: Dict[str, Any]) -> TomtomTrafficDensityJob:
+        return cls(
+            job_name=dict_object['jobName'],
+            network=TomtomNetwork.from_dict(dict_object['network']),
+            date_range=TomtomDateRange.from_dict(dict_object['dateRange']),
+            time_sets=[TomtomTimeSet.from_dict(t) for t in dict_object['timeSets']],
+            distance_unit=dict_object['distanceUnit'],
+            accept_mode=dict_object['acceptMode'],
+            map_version=dict_object['mapVersion'],
+            average_sample_size_threshold=None if 'averageSampleSizeThreshold' not in dict_object else dict_object[
+                'averageSampleSizeThreshold']
+        )