/usr/lib/python2.7/dist-packages/overpass/api.py is in python-overpass 0.5.6-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | import requests
import json
import geojson
from .errors import (OverpassSyntaxError, TimeoutError, MultipleRequestsError,
ServerLoadError, UnknownOverpassError, ServerRuntimeError)
class API(object):
"""A simple Python wrapper for the OpenStreetMap Overpass API"""
SUPPORTED_FORMATS = ["geojson", "json", "xml"]
# defaults for the API class
_timeout = 25 # seconds
_endpoint = "https://overpass-api.de/api/interpreter"
_debug = False
_QUERY_TEMPLATE = "[out:{out}];{query}out {verbosity};"
_GEOJSON_QUERY_TEMPLATE = "[out:json];{query}out body geom;"
def __init__(self, *args, **kwargs):
self.endpoint = kwargs.get("endpoint", self._endpoint)
self.timeout = kwargs.get("timeout", self._timeout)
self.debug = kwargs.get("debug", self._debug)
self._status = None
if self.debug:
import httplib
import logging
httplib.HTTPConnection.debuglevel = 1
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
def Get(self, query, responseformat="geojson", verbosity="body", build=True):
"""Pass in an Overpass query in Overpass QL"""
# Construct full Overpass query
if build:
full_query = self._ConstructQLQuery(query, responseformat=responseformat, verbosity=verbosity)
else:
full_query = query
# Get the response from Overpass
raw_response = self._GetFromOverpass(full_query)
if responseformat == "xml" or responseformat.startswith("csv"):
return raw_response
response = json.loads(raw_response)
# Check for valid answer from Overpass. A valid answer contains an 'elements' key at the root level.
if "elements" not in response:
raise UnknownOverpassError("Received an invalid answer from Overpass.")
# If there is a 'remark' key, it spells trouble.
overpass_remark = response.get('remark', None)
if overpass_remark and overpass_remark.startswith('runtime error'):
raise ServerRuntimeError(overpass_remark)
if responseformat is not "geojson":
return response
# construct geojson
return self._asGeoJSON(response["elements"])
def Search(self, feature_type, regex=False):
"""Search for something."""
raise NotImplementedError()
def _ConstructQLQuery(self, userquery, responseformat, verbosity):
raw_query = str(userquery)
if not raw_query.endswith(";"):
raw_query += ";"
if responseformat == "geojson":
template = self._GEOJSON_QUERY_TEMPLATE
complete_query = template.format(query=raw_query, verbosity=verbosity)
else:
template = self._QUERY_TEMPLATE
complete_query = template.format(query=raw_query, out=responseformat, verbosity=verbosity)
if self.debug:
print(complete_query)
return complete_query
def _GetFromOverpass(self, query):
"""This sends the API request to the Overpass instance and
returns the raw result, or an error."""
payload = {"data": query}
try:
r = requests.post(
self.endpoint,
data=payload,
timeout=self.timeout,
headers={'Accept-Charset': 'utf-8;q=0.7,*;q=0.7'}
)
except requests.exceptions.Timeout:
raise TimeoutError(self._timeout)
self._status = r.status_code
if self._status != 200:
if self._status == 400:
raise OverpassSyntaxError(query)
elif self._status == 429:
raise MultipleRequestsError()
elif self._status == 504:
raise ServerLoadError(self._timeout)
raise UnknownOverpassError(
"The request returned status code {code}".format(
code=self._status
)
)
else:
r.encoding = 'utf-8'
return r.text
def _asGeoJSON(self, elements):
features = []
for elem in elements:
elem_type = elem["type"]
if elem_type == "node":
geometry = geojson.Point((elem["lon"], elem["lat"]))
elif elem_type == "way":
points = []
for coords in elem["geometry"]:
points.append((coords["lon"], coords["lat"]))
geometry = geojson.LineString(points)
else:
continue
feature = geojson.Feature(
id=elem["id"],
geometry=geometry,
properties=elem.get("tags"))
features.append(feature)
return geojson.FeatureCollection(features)
|