Python Detour Batch Mode Demo
If you want to create large amounts of Detour segments automatically, you should use the batch mode instead of adding all segments with separate requests.
This example shows how you can add all roadworks from an Open Data dataset published by the city of Cologne into a Detour file.
To do this, the data sets are parsed, the roadworks are filtered, matched to the road segment data, attributes such as name and time periods are added, and in the end all segments are added to the Detour file.
Loading and parsing the roadworks
The data is published as a JSON file containing an array of features.
A feature looks like this:
{
"attributes": {
"objectid": 979225,
"name": "Luxemburger Straße (Sülz)",
"datum_von": 1669104000000,
"datum_bis": 1677247200000,
"link": "/leben-in-koeln/verkehr/verkehrskalender/stoerungen/17249/index.html",
"typ": 3,
"anzeige": 1,
"beschreibung": "Die Luxemburger Straße ist zwischen Universitätsstraße und Greinstraße in Fahrtrichtung stadtauswärts eingeengt."
},
"geometry": {
"x": 6.9338580266526977,
"y": 50.922109820387853
}
}
The highlighted lines (name, date from, date to, type and coordinate) should be imported to the Detour file.
The value typ indicates what kind of traffic information it is. According to the metadata description of the dataset, features of type 2 are roadworks that lead to road closure and features of type 3 are other roadworks.
This code loads the data from the API, parses the JSON document and filters all features with type 2 (roadworks with blocked roads)
and type 2 (other roadworks):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34 | import json
import requests
URL_ROADWORKS = 'https://geoportal.stadt-koeln.de/arcgis/rest/services/verkehr/verkehrskalender/MapServer/0/query' \
'?where=objectid%20is%20not%20null&text=&objectIds=&time=&geometry=&geometryType=esriGeometryEnvelope' \
'&inSR=&spatialRel=esriSpatialRelIntersects&distance=&units=esriSRUnit_Foot&relationParam=' \
'&outFields=%2A&returnGeometry=true&returnTrueCurves=false&maxAllowableOffset=&geometryPrecision=' \
'&outSR=4326&havingClause=&returnIdsOnly=false&returnCountOnly=false&orderByFields=' \
'&groupByFieldsForStatistics=&outStatistics=&returnZ=false&returnM=false&gdbVersion=&historicMoment=' \
'&returnDistinctValues=false&resultOffset=&resultRecordCount=&returnExtentOnly=false' \
'&datumTransformation=¶meterValues=&rangeValues=&quantizationParameters=' \
'&featureEncoding=esriDefault&f=pjson'
ROADWORKS_BLOCKED = 2
ROADWORKS = 3
def load_features():
response_text = requests.get(URL_ROADWORKS).text
data_json = json.loads(response_text)
return data_json['features']
def get_features_of_type(features, feature_type):
result = []
for feature in features:
if feature['attributes']['typ'] == feature_type:
result.append(feature)
return result
features = load_features()
roadworks_blocked = get_features_of_type(features, ROADWORKS_BLOCKED)
roadworks = get_features_of_type(features, ROADWORKS)
|
Querying Detour attributes for the roadworks
For all features in roadworks_blocked a Detour segment with the attribute block will be created. The features in roadworks will get the attribute penalize.
This code queries the attributes from the MapTrip Server API:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 | URL_SERVER_API = 'https://api.maptrip.de/'
TOKEN = '<insert your MapTrip Server API token here>'
headers = {
"Accept": "application/json",
"Authorization": "Bearer {token}".format(token=TOKEN)
}
def get_attributes(path):
response = requests.get(URL_SERVER_API + path, headers=headers)
if response.status_code == 200:
return json.loads(response.text)
else:
print(f'Failed to query attribute: {response}')
exit(0)
attributes_blocked = get_attributes('v1/detour/attribute/block?direction=both')
attributes_penalized = get_attributes('v1/detour/attribute/penalize')
|
Matching the roadworks to Detour segments
To turn the features into segments, they must be matched to the road data using the endpoint [POST] /detour/match.
The function add_features_to_segments matches the coordinates to segments, adds some attributes and adds the result to segments:
1
2
3
4
5
6
7
8
9
10
11
12
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 | import pytz
import urllib.parse
from datetime import datetime
PROVIDER = 'tomtom'
timezone = pytz.timezone('Europe/Berlin')
def match_feature_to_segment(feature):
coordinate = [{
'lat': feature['geometry']['y'],
'lon': feature['geometry']['x']
}]
response = requests.post(URL_SERVER_API + 'v1/detour/match?provider=' + PROVIDER, json=coordinate, headers=headers)
if response.status_code == 200:
return json.loads(response.text)
else:
return None
def add_features_to_segments(features_to_add, attributes):
for feature in features_to_add:
match = match_feature_to_segment(feature)
if match is not None:
segment_attributes = attributes.copy()
segment_attributes.append({'name': 'description', 'value': feature['attributes']['name']})
date_from = datetime.fromtimestamp(feature['attributes']['datum_von'] / 1000, timezone).isoformat()
date_to = datetime.fromtimestamp(feature['attributes']['datum_bis'] / 1000, timezone).isoformat()
att_from = get_attributes('v1/detour/attribute/timedomain/start?date=' + urllib.parse.quote(date_from))
att_to = get_attributes('v1/detour/attribute/timedomain/end?date=' + urllib.parse.quote(date_to))
segment_attributes.append(att_from)
segment_attributes.append(att_to)
segment = {
'coordinates': match['coordinates'],
'openlr': match['openlr'],
'attributes': segment_attributes
}
segments.append(segment)
segments = []
add_features_to_segments(roadworks_blocked, attributes_blocked)
add_features_to_segments(roadworks, attributes_penalized)
|
Removing existing segments from a Detour file
If you want to do a batch insert on a regular basis, you should remove the old segments
before inserting them again. This can be done using the endpoint [DELETE] /detour/file/{fileid}/segments:
| DETOUR_FILE_ID = '<insert your file ID here>'
def delete_all_segments():
url = URL_SERVER_API + f'v1/detour/file/{DETOUR_FILE_ID}/segments/?provider={PROVIDER}'
response = requests.delete(url, headers=headers)
return response.status_code == 200
if delete_all_segments():
print(f'Deleted all existing segments from detour file')
|
Adding the segments to a Detour file
Using the endpoint [POST] /detour/file/{fileid}/segments the segments are are added to the Detour file:
| def add_segments(segments):
url = URL_SERVER_API + f'v1/detour/file/{DETOUR_FILE_ID}/segments/?provider={PROVIDER}'
response = requests.post(url, json=segments, headers=headers)
return response.status_code == 200
if add_segments(segments):
print(f'Successfully added {len(roadworks_blocked)} blocked and {len(roadworks)} normal roadworks to detour file')
else:
print('Failed')
|
Complete example
Here is the complete code of the previous sections:
1
2
3
4
5
6
7
8
9
10
11
12
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 | import json
import requests
import pytz
import urllib.parse
from datetime import datetime
PROVIDER = 'tomtom'
DETOUR_FILE_ID = '<insert your file ID here>'
URL_SERVER_API = 'https://api.maptrip.de/'
TOKEN = '<insert your MapTrip Server API token here>'
URL_ROADWORKS = 'https://geoportal.stadt-koeln.de/arcgis/rest/services/verkehr/verkehrskalender/MapServer/0/query' \
'?where=objectid%20is%20not%20null&text=&objectIds=&time=&geometry=&geometryType=esriGeometryEnvelope' \
'&inSR=&spatialRel=esriSpatialRelIntersects&distance=&units=esriSRUnit_Foot&relationParam=' \
'&outFields=%2A&returnGeometry=true&returnTrueCurves=false&maxAllowableOffset=&geometryPrecision=' \
'&outSR=4326&havingClause=&returnIdsOnly=false&returnCountOnly=false&orderByFields=' \
'&groupByFieldsForStatistics=&outStatistics=&returnZ=false&returnM=false&gdbVersion=&historicMoment=' \
'&returnDistinctValues=false&resultOffset=&resultRecordCount=&returnExtentOnly=false' \
'&datumTransformation=¶meterValues=&rangeValues=&quantizationParameters=' \
'&featureEncoding=esriDefault&f=pjson'
ROADWORKS_BLOCKED = 2
ROADWORKS = 3
timezone = pytz.timezone('Europe/Berlin')
headers = {
"Accept": "application/json",
"Authorization": "Bearer {token}".format(token=TOKEN)
}
def load_features():
response_text = requests.get(URL_ROADWORKS).text
data_json = json.loads(response_text)
return data_json['features']
def get_features_of_type(features, feature_type):
result = []
for feature in features:
if feature['attributes']['typ'] == feature_type:
result.append(feature)
return result
def get_attributes(path):
response = requests.get(URL_SERVER_API + path, headers=headers)
if response.status_code == 200:
return json.loads(response.text)
else:
print(f'Failed to query attribute: {response}')
exit(0)
def match_feature_to_segment(feature):
coordinate = [{
'lat': feature['geometry']['y'],
'lon': feature['geometry']['x']
}]
response = requests.post(URL_SERVER_API + 'v1/detour/match?provider=' + PROVIDER, json=coordinate, headers=headers)
if response.status_code == 200:
return json.loads(response.text)
else:
return None
def delete_all_segments():
url = URL_SERVER_API + f'v1/detour/file/{DETOUR_FILE_ID}/segments/?provider={PROVIDER}'
response = requests.delete(url, headers=headers)
return response.status_code == 200
def add_features_to_segments(features_to_add, attributes):
for feature in features_to_add:
match = match_feature_to_segment(feature)
if match is not None:
segment_attributes = attributes.copy()
segment_attributes.append({'name': 'description', 'value': feature['attributes']['name']})
date_from = datetime.fromtimestamp(feature['attributes']['datum_von'] / 1000, timezone).isoformat()
date_to = datetime.fromtimestamp(feature['attributes']['datum_bis'] / 1000, timezone).isoformat()
att_from = get_attributes('v1/detour/attribute/timedomain/start?date=' + urllib.parse.quote(date_from))
att_to = get_attributes('v1/detour/attribute/timedomain/end?date=' + urllib.parse.quote(date_to))
segment_attributes.append(att_from)
segment_attributes.append(att_to)
segment = {
'coordinates': match['coordinates'],
'openlr': match['openlr'],
'attributes': segment_attributes
}
segments.append(segment)
def add_segments(segments):
url = URL_SERVER_API + f'v1/detour/file/{DETOUR_FILE_ID}/segments/?provider={PROVIDER}'
response = requests.post(url, json=segments, headers=headers)
return response.status_code == 200
features = load_features()
segments = []
roadworks_blocked = get_features_of_type(features, ROADWORKS_BLOCKED)
attributes_blocked = get_attributes('v1/detour/attribute/block?direction=both')
add_features_to_segments(roadworks_blocked, attributes_blocked)
roadworks = get_features_of_type(features, ROADWORKS)
attributes_penalized = get_attributes('v1/detour/attribute/penalize')
add_features_to_segments(roadworks, attributes_penalized)
if delete_all_segments():
print(f'Deleted all existing segments from detour file')
if add_segments(segments):
print(f'Successfully added {len(roadworks_blocked)} blocked and {len(roadworks)} normal roadworks to detour file')
else:
print('Failed')
|
When viewed in the Detour Editor, you can see blocked and penalized segments which have the name and date from the JSON dataset:
