From de916f7305ba4934c490297a24e91f880b8a1b1a Mon Sep 17 00:00:00 2001 From: Guillaume Latour Date: Thu, 27 Jun 2024 15:06:08 +0200 Subject: [PATCH] feat: add maps feature + ruff format --- pyproject.toml | 8 +- src/tomtom_api/cli.py | 2 +- src/tomtom_api/client.py | 274 +++++++++++------- src/tomtom_api/priority_queue/cli.py | 2 - .../priority_queue/models/daemon.py | 3 +- .../priority_queue/models/database.py | 2 +- .../traffic_stats/models/geospatial.py | 4 +- .../traffic_stats/models/jobs/base.py | 2 +- .../traffic_stats/models/jobs/route.py | 4 +- src/tomtom_api/traffic_stats/models/time.py | 2 +- src/tomtom_api/utils/time.py | 2 +- 11 files changed, 189 insertions(+), 116 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 094ca90..238a465 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,22 +13,22 @@ dependencies = [ 'pandas', 'pyarrow', 'tabulate', - 'geojson' + 'geojson', ] description = "API client allowing to interact with the tomtom REST services." maintainers = [ - { name = "Guillaume Latour", email = "guillaume.latour@macq.eu" } + { name = "Guillaume Latour", email = "guillaume.latour@macq.eu" }, ] name = "tomtom_api" readme = "readme.md" requires-python = ">=3.8" -version = "2023.5.24" +version = "2024.6.27" [project.urls] repository = "https://stash.macq.eu/projects/RDDS/repos/tomtom-api" [project.optional-dependencies] -dev = ['autopep8', 'pytest', 'pdoc'] +dev = ['ruff', 'pytest'] pyproj = ['pyproj'] [project.scripts] diff --git a/src/tomtom_api/cli.py b/src/tomtom_api/cli.py index 5ac7cd1..76c21a5 100644 --- a/src/tomtom_api/cli.py +++ b/src/tomtom_api/cli.py @@ -43,7 +43,7 @@ def client(): @click.option('-t', '--job-type', type=click.Choice(['routeanalysis', 'areaanalysis', 'trafficdensity']), multiple=True, help="The type of jobs.") @click.option('-s', '--state', type=click.Choice([n.name for n in TomtomJobState]), multiple=True, - help=f"The current state of jobs.") + help="The current state of jobs.") def list_all_job_info( base_url: Optional[str] = None, version: Optional[int] = None, diff --git a/src/tomtom_api/client.py b/src/tomtom_api/client.py index 2d5a022..2489689 100644 --- a/src/tomtom_api/client.py +++ b/src/tomtom_api/client.py @@ -1,24 +1,32 @@ import datetime as dt +from shapely import geometry as geo import json from typing import List, Literal, Optional, Union import requests from dataclass_wizard.errors import MissingFields +import geojson from tomtom_api import config, log -from tomtom_api.traffic_stats.models.geospatial import (TomtomNetwork, - TomtomRoad) +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.responses import ( - JsonIpResponse, TomtomErrorResponse, TomtomJobInfo, TomtomResponseAnalysis, - TomtomResponseRouteFound, TomtomResponseSearchJobs, TomtomResponseStatus) -from tomtom_api.traffic_stats.models.status import (TomtomJobState, - TomtomReportType) + JsonIpResponse, + TomtomErrorResponse, + TomtomJobInfo, + TomtomResponseAnalysis, + TomtomResponseRouteFound, + TomtomResponseSearchJobs, + TomtomResponseStatus, +) +from tomtom_api.traffic_stats.models.status import TomtomJobState, TomtomReportType from tomtom_api.traffic_stats.models.time import TomtomDateRange, TomtomTimeSet from tomtom_api.utils.time import date_as_str -SubmittedTomtomJob = Union[int, TomtomResponseAnalysis, TomtomResponseStatus, TomtomJobInfo] +SubmittedTomtomJob = Union[ + int, TomtomResponseAnalysis, TomtomResponseStatus, TomtomJobInfo +] class TomtomClient: @@ -35,7 +43,7 @@ def __init__( proxy_ip: Optional[str] = None, proxy_port: Optional[int] = None, proxy_username: Optional[str] = None, - proxy_password: Optional[str] = None + proxy_password: Optional[str] = None, ): self.key: str = key or config.api.key self.version: int = version or config.api.version @@ -47,25 +55,35 @@ def __init__( proxy_username = proxy_username or config.api.proxy.username proxy_password = proxy_password or config.api.proxy.password - tab = [proxy_ip is None, proxy_port is None, proxy_username is None, proxy_password is None] + tab = [ + proxy_ip is None, + proxy_port is None, + proxy_username is None, + proxy_password is None, + ] if not all([v == tab[0] for v in tab[1:]]): - raise ValueError('Some of the proxy information were given, but not all of them.\n' - 'Please check that you are defining all the proxy related variables.') - - self.proxy_url: Optional[str] = None if proxy_ip is None \ + raise ValueError( + "Some of the proxy information were given, but not all of them.\n" + "Please check that you are defining all the proxy related variables." + ) + + self.proxy_url: Optional[str] = ( + None + if proxy_ip is None else f"http://{proxy_username}:{proxy_password}@{proxy_ip}:{proxy_port}" + ) if self.key is None: - raise ValueError('The client cannot be initialized without an API key.') + raise ValueError("The client cannot be initialized without an API key.") if self.version is None: - raise ValueError('The client cannot be initialized without a version.') + raise ValueError("The client cannot be initialized without a version.") if self.base_url is None: - raise ValueError('The client cannot be initialized without a base url.') + raise ValueError("The client cannot be initialized without a base url.") def __del__(self): try: self.session.close() - except: + except Exception: pass def request(self, *args, **kwargs) -> requests.Response: @@ -82,22 +100,32 @@ def request(self, *args, **kwargs) -> requests.Response: Exception If the user was not allowed to perform the query (403 Forbidden) """ - proxies = None if self.proxy_url is None else {protocol: self.proxy_url for protocol in ['http', 'https']} - headers = {'content-type': 'application/json'} - r = self.session.request(*args, proxies=proxies, headers=headers, allow_redirects=True, **kwargs) - log.debug(f'request: {r.text}') + proxies = ( + None + if self.proxy_url is None + else {protocol: self.proxy_url for protocol in ["http", "https"]} + ) + headers = {"content-type": "application/json"} + r = self.session.request( + *args, proxies=proxies, headers=headers, allow_redirects=True, **kwargs + ) + log.debug(f"request: {r.text}") if r.status_code == 403: log.error(r.text) # todo: create custom exception for forbidden - raise Exception('The access is forbidden. Maybe you should check if the correct API key is loaded.') + raise Exception( + "The access is forbidden. Maybe you should check if the correct API key is loaded." + ) if r.status_code == 400: try: error = TomtomErrorResponse.from_dict(r.json()) for message in error.messages: - log.error(f"{message.error} ({message.field}):\n{message.rejected_value}") - except MissingFields as e: + log.error( + f"{message.error} ({message.field}):\n{message.rejected_value}" + ) + except MissingFields: error = TomtomResponseAnalysis.from_dict(r.json()) for message in error.messages: log.error(f"Job {error.job_id}: {message}") @@ -105,20 +133,22 @@ def request(self, *args, **kwargs) -> requests.Response: # we can just return the original request.response return r - raise Exception(f'The given request generated an error (HTTP400). Please check logs for more info.') + raise Exception( + "The given request generated an error (HTTP400). Please check logs for more info." + ) return r def route_analysis( self, job_name: str, - distance_unit: Literal['KILOMETERS', 'MILES'], + distance_unit: Literal["KILOMETERS", "MILES"], roads: List[TomtomRoad], date_ranges: List[TomtomDateRange], time_sets: List[TomtomTimeSet], - accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO', - map_version: Optional[float] = 2016.12, - average_sample_size_threshold: Optional[int] = None + accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO", + map_version: Optional[float] = 2023.03, + average_sample_size_threshold: Optional[int] = None, ) -> TomtomResponseAnalysis: """ The Traffic Stats API Route Analysis service calculates the statistics for a defined route between an origin @@ -126,26 +156,28 @@ def route_analysis( Documentation: https://developer.tomtom.com/traffic-stats/documentation/api/route-analysis """ - job = TomtomRouteJob(job_name=job_name, - distance_unit=distance_unit, - routes=roads, - date_ranges=date_ranges, - time_sets=time_sets, - accept_mode=accept_mode, - map_version=map_version, - average_sample_size_threshold=average_sample_size_threshold) + job = TomtomRouteJob( + job_name=job_name, + distance_unit=distance_unit, + routes=roads, + date_ranges=date_ranges, + time_sets=time_sets, + accept_mode=accept_mode, + map_version=map_version, + average_sample_size_threshold=average_sample_size_threshold, + ) return self.post_job_route_analysis(job) def area_analysis( self, job_name: str, - distance_unit: Literal['KILOMETERS', 'MILES'], + distance_unit: Literal["KILOMETERS", "MILES"], network: TomtomNetwork, date_range: TomtomDateRange, time_sets: List[TomtomTimeSet], - accept_mode: Optional[Literal['AUTO', 'MANUAL']] = 'AUTO', + accept_mode: Optional[Literal["AUTO", "MANUAL"]] = "AUTO", map_version: Optional[float] = 2023.03, - average_sample_size_threshold: Optional[int] = None + average_sample_size_threshold: Optional[int] = None, ) -> TomtomResponseAnalysis: job = TomtomAreaJob( job_name=job_name, @@ -155,15 +187,15 @@ def area_analysis( time_sets=time_sets, accept_mode=accept_mode, map_version=map_version, - average_sample_size_threshold=average_sample_size_threshold + average_sample_size_threshold=average_sample_size_threshold, ) return self.post_job_area_analysis(job) def find_route( self, road: TomtomRoad, - map_version: Optional[str] = '2016.12', - map_type: Optional[str] = 'DSEG_NOSPLIT' + map_version: Optional[str] = "2016.12", + map_type: Optional[str] = "DSEG_NOSPLIT", ) -> TomtomResponseRouteFound: """ This API performs a check to see whether the road can be found with the given informations. @@ -182,30 +214,27 @@ def find_route( TomtomResponseRouteFound The response of the API. """ - url = f'https://{self.base_url}/traffic/trafficstats/route?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/route?key={self.key}" data = road.to_find_route_api_object(map_version, map_type) - response = self.request('post', url, data=json.dumps(data)) + response = self.request("post", url, data=json.dumps(data)) tomtom_response = TomtomResponseRouteFound.from_dict(response.json()) return tomtom_response def post_job_route_analysis(self, job: TomtomRouteJob) -> TomtomResponseAnalysis: - url = f'https://{self.base_url}/traffic/trafficstats/routeanalysis/{self.version}?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/routeanalysis/{self.version}?key={self.key}" - request_response = self.request('post', url, data=json.dumps(job.to_dict())) + request_response = self.request("post", url, data=json.dumps(job.to_dict())) tomtom_response = TomtomResponseAnalysis.from_dict(request_response.json()) return tomtom_response def post_job_area_analysis(self, job: TomtomAreaJob) -> TomtomResponseAnalysis: - url = f'https://{self.base_url}/traffic/trafficstats/areaanalysis/{self.version}?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/areaanalysis/{self.version}?key={self.key}" - request_response = self.request('post', url, data=json.dumps(job.to_dict())) + 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: + def status(self, job_id: int) -> TomtomResponseStatus: """When a job has been initiated via the API request it is possible to check the status. Documentation: @@ -223,12 +252,40 @@ def status( TomtomResponseStatus The requested information. """ - url = f'https://{self.base_url}/traffic/trafficstats/status/{self.version}/{job_id}?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/status/{self.version}/{job_id}?key={self.key}" - request_response = self.request('get', url) + request_response = self.request("get", url) tomtom_response = TomtomResponseStatus.from_dict(request_response.json()) return tomtom_response + def available_maps( + self, + geometry: Union[geo.Polygon, geo.MultiPolygon], + start: dt.date, + end: dt.date, + ) -> List[str]: + + url = f"https://{self.base_url}/traffic/trafficstats/maps/{self.version}?key={self.key}" + day_format = "%Y-%m-%d" + data = { + "geometry": geojson.Feature(geometry=geometry, properties={}).geometry, + "dateRange": { + "from": start.strftime(day_format), + "to": end.strftime(day_format), + }, + } + request_response = self.request( + "post", + url, + data=json.dumps(data), + ) + + if request_response.status_code == 400: + messages = request_response.json()["messages"] + message = "\n".join(messages) + raise Exception(message) + return request_response.json()["maps"] + def search_jobs( self, page_index: Optional[int] = None, @@ -240,7 +297,7 @@ def search_jobs( name: Optional[str] = None, job_id: Optional[int] = None, job_type: Optional[Union[TomtomReportType, List[TomtomReportType]]] = None, - state: Optional[Union[TomtomJobState, List[TomtomJobState]]] = None + state: Optional[Union[TomtomJobState, List[TomtomJobState]]] = None, ) -> TomtomResponseSearchJobs: """Fetch information about the jobs you have. You can either get them all or use some filters. @@ -275,15 +332,17 @@ def search_jobs( TomtomResponseSearchJob The response provided by the tomtom API """ - url = f'https://{self.base_url}/traffic/trafficstats/job/search/{self.version}?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/job/search/{self.version}?key={self.key}" if page_index is not None and page_index < 0: - raise ValueError(f'page_index must be greater or equal to zero ({page_index} given)') + raise ValueError( + f"page_index must be greater or equal to zero ({page_index} given)" + ) if per_page is not None and per_page <= 0: - raise ValueError(f'per_page must be greater than zero ({per_page} given)') + raise ValueError(f"per_page must be greater than zero ({per_page} given)") - date_fmt = '%Y-%m-%d' + date_fmt = "%Y-%m-%d" # expand parameters that can be lists job_type = job_type if isinstance(job_type, list) else [job_type] @@ -292,20 +351,36 @@ def search_jobs( state = [s for s in state if s is not None] params = { - 'pageIndex': page_index, - 'perPage': per_page, - 'createdAfter': None if created_after is None else date_as_str(created_after, date_fmt), - 'createdBefore': None if created_before is None else date_as_str(created_before, date_fmt), - 'completedAfter': None if completed_after is None else date_as_str(completed_after, date_fmt), - 'completedBefore': None if completed_before is None else date_as_str(completed_before, date_fmt), - 'name': name, - 'id': job_id, - 'type': None if len(job_type) < 1 else ','.join([j.value for j in job_type]), - 'state': None if len(state) < 1 else ','.join([s.name for s in state]) + "pageIndex": page_index, + "perPage": per_page, + "createdAfter": ( + None if created_after is None else date_as_str(created_after, date_fmt) + ), + "createdBefore": ( + None + if created_before is None + else date_as_str(created_before, date_fmt) + ), + "completedAfter": ( + None + if completed_after is None + else date_as_str(completed_after, date_fmt) + ), + "completedBefore": ( + None + if completed_before is None + else date_as_str(completed_before, date_fmt) + ), + "name": name, + "id": job_id, + "type": ( + None if len(job_type) < 1 else ",".join([j.value for j in job_type]) + ), + "state": None if len(state) < 1 else ",".join([s.name for s in state]), } params = {k: v for k, v in params.items() if v is not None} - request_response = self.request('get', url, params=params) + request_response = self.request("get", url, params=params) return TomtomResponseSearchJobs.from_dict(request_response.json()) def delete_job(self, job: Union[SubmittedTomtomJob, List[SubmittedTomtomJob]]): @@ -314,9 +389,9 @@ def delete_job(self, job: Union[SubmittedTomtomJob, List[SubmittedTomtomJob]]): job_id = job if isinstance(job, int) else job.job_id - url = f'https://{self.base_url}/traffic/trafficstats/reports/{job_id}/?key={self.key}' + url = f"https://{self.base_url}/traffic/trafficstats/reports/{job_id}/?key={self.key}" - response = self.request('delete', url) + response = self.request("delete", url) return response @@ -327,13 +402,15 @@ def cancel_job(self, job: Union[SubmittedTomtomJob, List[SubmittedTomtomJob]]): job_id = job if isinstance(job, int) else job.job_id url = f"https://{self.base_url}/traffic/trafficstats/status/{self.version}/{job_id}/cancel?key={self.key}" - response = self.request('post', url) + response = self.request("post", url) return response def check_ip(self, use_proxy: bool = False) -> JsonIpResponse: if use_proxy and self.proxy_url is None: - raise Exception('Trying to use a proxy while it is not configured.\n' - 'Try to setup the client with proxy values or look at your environment variables') + raise Exception( + "Trying to use a proxy while it is not configured.\n" + "Try to setup the client with proxy values or look at your environment variables" + ) url = "https://jsonip.com/" @@ -343,7 +420,7 @@ def check_ip(self, use_proxy: bool = False) -> JsonIpResponse: self.proxy_url = None # perform the query - response = self.request('get', url) + response = self.request("get", url) # do not forget to reset the self.proxy_url self.proxy_url = tmp_proxy @@ -357,43 +434,44 @@ def __init__(self, *args, **kwargs): super().__init__(**kwargs) def request(self, *args, **kwargs): - raise Exception('This client is not meant to send requests.') + raise Exception("This client is not meant to send requests.") def route_analysis(self, *args, **kwargs) -> TomtomResponseAnalysis: - return TomtomResponseAnalysis('OK', ['This is a dummy response'], -1) + return TomtomResponseAnalysis("OK", ["This is a dummy response"], -1) def post_job_route_analysis(self, *args, **kwargs) -> TomtomResponseAnalysis: - return TomtomResponseAnalysis(response_status='OK', - messages=['This is a dummy response'], - job_id=1) + return TomtomResponseAnalysis( + response_status="OK", messages=["This is a dummy response"], job_id=1 + ) def post_job_area_analysis(self, *args, **kwargs) -> TomtomResponseAnalysis: - return TomtomResponseAnalysis(response_status='OK', - messages=['This is a dummy response'], - job_id=1) + return TomtomResponseAnalysis( + response_status="OK", messages=["This is a dummy response"], job_id=1 + ) def status(self, *args, **kwargs) -> TomtomResponseStatus: - return TomtomResponseStatus(-1, TomtomJobState.DONE, 'OK', ['https://google.be']) + return TomtomResponseStatus( + -1, TomtomJobState.DONE, "OK", ["https://google.be"] + ) def search_jobs(self, *args, **kwargs) -> TomtomResponseSearchJobs: - from tomtom_api.traffic_stats.models.responses import (Pageable, Sort, - TomtomJobInfo) + from tomtom_api.traffic_stats.models.responses import ( + Pageable, + Sort, + TomtomJobInfo, + ) + job_info = TomtomJobInfo( - name='dummy job', + name="dummy job", created_at=dt.datetime.now(), state=TomtomJobState.DONE, job_id=1, completed_at=dt.datetime.now(), - job_type='dummy' + job_type="dummy", ) sort = Sort(is_sorted=True, is_unsorted=False, is_empty=False) pageable = Pageable( - sort=sort, - page_size=1, - page_number=1, - offset=1, - paged=True, - unpaged=False + sort=sort, page_size=1, page_number=1, offset=1, paged=True, unpaged=False ) return TomtomResponseSearchJobs( content=[job_info], @@ -406,8 +484,8 @@ def search_jobs(self, *args, **kwargs) -> TomtomResponseSearchJobs: number_of_elements=1, size=1, number=1, - empty=False + empty=False, ) def check_ip(self, *args, **kwargs) -> JsonIpResponse: - raise Exception('This client is not meant to send requests.') + raise Exception("This client is not meant to send requests.") diff --git a/src/tomtom_api/priority_queue/cli.py b/src/tomtom_api/priority_queue/cli.py index 349d806..3c49e6c 100644 --- a/src/tomtom_api/priority_queue/cli.py +++ b/src/tomtom_api/priority_queue/cli.py @@ -3,7 +3,6 @@ import click -from tomtom_api import config from tomtom_api.priority_queue.lib import (pretty_print_queue, priority_queue_add_job, priority_queue_clean_folder, @@ -12,7 +11,6 @@ priority_queue_update_job) from tomtom_api.priority_queue.models.daemon import PriorityQueueDaemon from tomtom_api.priority_queue.models.database import PriorityQueueDB -from tomtom_api.priority_queue.models.queue import QueueItem from tomtom_api.priority_queue.models.status import QueueItemStatus from tomtom_api.traffic_stats.models.jobs.route import TomtomRouteJob diff --git a/src/tomtom_api/priority_queue/models/daemon.py b/src/tomtom_api/priority_queue/models/daemon.py index 8188e97..f958302 100644 --- a/src/tomtom_api/priority_queue/models/daemon.py +++ b/src/tomtom_api/priority_queue/models/daemon.py @@ -9,8 +9,7 @@ from tomtom_api.priority_queue.models.queue import QueueItem from tomtom_api.priority_queue.models.status import QueueItemStatus from tomtom_api.traffic_stats import N_CONCURRENT_JOB_IN_PROGRESS -from tomtom_api.traffic_stats.models.status import (TomtomJobState, - TomtomReportType) +from tomtom_api.traffic_stats.models.status import (TomtomJobState) from tomtom_api.utils.daemon import Daemon DAEMON_LOG_FILE = config.path.home / 'daemon.log' diff --git a/src/tomtom_api/priority_queue/models/database.py b/src/tomtom_api/priority_queue/models/database.py index 5be107d..2050903 100644 --- a/src/tomtom_api/priority_queue/models/database.py +++ b/src/tomtom_api/priority_queue/models/database.py @@ -149,7 +149,7 @@ def get_filtered_items( uid, name, priority, status = uid or [], name or [], priority or [], status or [] # uid filter - uid_filter_str = '' if len(uid) < 1 else f'(uid in @uid)' + uid_filter_str = '' if len(uid) < 1 else '(uid in @uid)' # name filter name_filters = [f'name.str.contains("{n}")' for n in name] diff --git a/src/tomtom_api/traffic_stats/models/geospatial.py b/src/tomtom_api/traffic_stats/models/geospatial.py index 9ac984c..0fe80de 100644 --- a/src/tomtom_api/traffic_stats/models/geospatial.py +++ b/src/tomtom_api/traffic_stats/models/geospatial.py @@ -182,7 +182,7 @@ def __init__( raise TooManyViaPointsException(f'The TomtomRoad {self.name} has too many via points.') if self.length() > MAX_ROAD_LENGTH: - raise RoadTooLongException(f'The road length is greater than the maximum allowed.') + raise RoadTooLongException('The road length is greater than the maximum allowed.') @classmethod def from_dict(cls, dict_object: Dict[str, Any], ignore_consecutive_points: bool = False) -> TomtomRoad: @@ -316,7 +316,7 @@ def __init__( self.geometry = geometry if self.area() > MAX_ROAD_LENGTH: - raise RoadTooLongException(f'The road length is greater than the maximum allowed.') + raise RoadTooLongException('The road length is greater than the maximum allowed.') @classmethod def from_dict(cls, dict_object: Dict[str, Any]) -> TomtomNetwork: diff --git a/src/tomtom_api/traffic_stats/models/jobs/base.py b/src/tomtom_api/traffic_stats/models/jobs/base.py index f7ab8d9..efc9322 100644 --- a/src/tomtom_api/traffic_stats/models/jobs/base.py +++ b/src/tomtom_api/traffic_stats/models/jobs/base.py @@ -3,7 +3,7 @@ import json from typing import Any, Dict, List, Literal, Optional -from tomtom_api.traffic_stats import MAX_DATE_RANGE_COUNT, MAX_TIME_SETS_COUNT +from tomtom_api.traffic_stats import MAX_TIME_SETS_COUNT from tomtom_api.traffic_stats.models.time import TomtomDateRange, TomtomTimeSet diff --git a/src/tomtom_api/traffic_stats/models/jobs/route.py b/src/tomtom_api/traffic_stats/models/jobs/route.py index b077ff8..e548ce5 100644 --- a/src/tomtom_api/traffic_stats/models/jobs/route.py +++ b/src/tomtom_api/traffic_stats/models/jobs/route.py @@ -1,10 +1,8 @@ from __future__ import annotations -import json from typing import Any, Dict, List, Literal, Optional -from tomtom_api.traffic_stats import (MAX_DATE_RANGE_COUNT, MAX_ROAD_COUNT, - MAX_TIME_SETS_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 diff --git a/src/tomtom_api/traffic_stats/models/time.py b/src/tomtom_api/traffic_stats/models/time.py index 927408d..13549b0 100644 --- a/src/tomtom_api/traffic_stats/models/time.py +++ b/src/tomtom_api/traffic_stats/models/time.py @@ -64,7 +64,7 @@ def __init__( if e < s: raise ValueError(f'The start date ({self.start}) is after the end date ({self.end})') if (e - s).days > MAX_DATE_RANGE_LENGTH: - raise ValueError(f'The given date range is greater than the maximum allowed.') + raise ValueError('The given date range is greater than the maximum allowed.') self.exclusions = None if exclusions is None else [date_as_str(d, date_fmt) for d in exclusions] self.excluded_days_of_week = None if excluded_days_of_week is None else dow(excluded_days_of_week) diff --git a/src/tomtom_api/utils/time.py b/src/tomtom_api/utils/time.py index 891658a..b9ed9e7 100644 --- a/src/tomtom_api/utils/time.py +++ b/src/tomtom_api/utils/time.py @@ -66,7 +66,7 @@ def dow(days: List[str]) -> List[str]: all_dow = {d.upper() for d in calendar.day_abbr} days = [d.upper() for d in days] if not all([d in all_dow for d in days]): - raise ValueError(f'At least one of the input in `days` is not a 3 first letters day word.') + raise ValueError('At least one of the input in `days` is not a 3 first letters day word.') return days