mirror of
https://github.com/enzet/map-machine.git
synced 2025-05-21 21:16:24 +02:00
Rename Röntgen to Map Machine.
This commit is contained in:
parent
38c4e00de3
commit
2bd89a6539
57 changed files with 252 additions and 236 deletions
70
map_machine/osm_getter.py
Normal file
70
map_machine/osm_getter.py
Normal file
|
@ -0,0 +1,70 @@
|
|||
"""
|
||||
Getting OpenStreetMap data from the web.
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import urllib3
|
||||
|
||||
from map_machine.boundary_box import BoundaryBox
|
||||
|
||||
__author__ = "Sergey Vartanov"
|
||||
__email__ = "me@enzet.ru"
|
||||
|
||||
SLEEP_TIME_BETWEEN_REQUESTS: float = 2.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class NetworkError(Exception):
|
||||
"""Failed network request."""
|
||||
|
||||
message: str
|
||||
|
||||
|
||||
def get_osm(
|
||||
boundary_box: BoundaryBox, cache_file_path: Path, to_update: bool = False
|
||||
) -> str:
|
||||
"""
|
||||
Download OSM data from the web or get if from the cache.
|
||||
|
||||
:param boundary_box: borders of the map part to download
|
||||
:param cache_file_path: cache file to store downloaded OSM data
|
||||
:param to_update: update cache files
|
||||
"""
|
||||
if not to_update and cache_file_path.is_file():
|
||||
with cache_file_path.open() as output_file:
|
||||
return output_file.read()
|
||||
|
||||
content: bytes = get_data(
|
||||
"https://api.openstreetmap.org/api/0.6/map",
|
||||
{"bbox": boundary_box.get_format()},
|
||||
)
|
||||
|
||||
with cache_file_path.open("bw+") as output_file:
|
||||
output_file.write(content)
|
||||
|
||||
return content.decode("utf-8")
|
||||
|
||||
|
||||
def get_data(address: str, parameters: dict[str, str]) -> bytes:
|
||||
"""
|
||||
Construct Internet page URL and get its descriptor.
|
||||
|
||||
:param address: URL without parameters
|
||||
:param parameters: URL parameters
|
||||
:return: connection descriptor
|
||||
"""
|
||||
logging.info(f"Getting {address}...")
|
||||
pool_manager: urllib3.PoolManager = urllib3.PoolManager()
|
||||
urllib3.disable_warnings()
|
||||
|
||||
try:
|
||||
result = pool_manager.request("GET", address, parameters)
|
||||
except urllib3.exceptions.MaxRetryError:
|
||||
raise NetworkError("Cannot download data: too many attempts.")
|
||||
|
||||
pool_manager.clear()
|
||||
time.sleep(SLEEP_TIME_BETWEEN_REQUESTS)
|
||||
return result.data
|
Loading…
Add table
Add a link
Reference in a new issue