pylib/geo
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
geo/osm.py

629 lines
24 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
"""
geo.osm (Open Streetmap)
========================
**Author:**
* Dirk Alders <sudo-dirk@mount-mockery.de>
**Description:**
This module is a submodule of :mod:`geo` and supports functions and classes for Open Streetmap.
**Contentlist:**
* :func:`geo.osm.landmark_link`
**Unittest:**
See also the :download:`unittest <../../geo/_testresults_/unittest.pdf>` documentation.
"""
MAP_STANDARD = 'N'
"""MAP definition for Standard Map"""
MAP_LOCAL_TRAFIC = 'TN'
"""MAP definition for Local Trafic Map"""
MAP_CYCLEMAP = 'CN'
"""MAP definition for Cyclemap"""
MAP_HUMANITARIAN = 'HN'
"""MAP definition for Humanitarian Map"""
def landmark_link(coord, zoom_level=13, map_code=MAP_STANDARD):
"""
:param coord: Target coordinate.
:type coord: geo.gps.coordinate
:param zoom_level: The zoom level of the map (see https://wiki.openstreetmap.org/wiki/Zoom_levels for more information)
:type zoom_level: int
:param map_code: One of the map_codes :class:`MAP_STANDARD`, :class:`MAP_LOCAL_TRAFIC`, :class:`MAP_CYCLEMAP`, :class:`MAP_HUMANITARIAN`.
:type map_code: str
:return: An openstreetmap-url for marking a position in a map.
:rtype: str
This Method generates an openstreetmap-url for marking a position in a map.
.. code-block:: python
>>> import geo
>>> gb = geo.gps.coordinate(lat=53.6908298,lon=12.1583252)
>>> geo.osm.landmark_link(gb)
'http://www.openstreetmap.org?mlat=53.690830&mlon=12.158325&zoom=13&layers=N'
"""
lon = coord[coord.LONGITUDE]
lat = coord[coord.LATITUDE]
#
if lon is not None and lat is not None:
link = 'http://www.openstreetmap.org?mlat=%(' + coord.LATITUDE + ')f&mlon=%(' + coord.LONGITUDE + ')f&zoom=%(zoom)d&layers=%(map)s'
return link % {coord.LATITUDE: lat,
coord.LONGITUDE: lon,
'zoom': zoom_level,
'map': map_code}
else:
return None
class map_spec(object):
def __init__(self, **kwargs):
coord1 = kwargs.get('coord1')
coord2 = kwargs.get('coord2')
zoom_level = kwargs.get('zoom_level')
x = kwargs.get('x')
y = kwargs.get('y')
def get_resolution(self):
pass
def get_coord_range(self):
pass
def get_map(self):
pass
def point_to_coord(self, xy):
pass
def coord_to_point(self, coord):
pass
class osm_map(object):
def __init__(self, **kwargs):
self.__map_spec__ = map_spec(kwargs)
map_code = kwargs.get('map_code')
'''
class get_from_web():
"""
Class to download images from web.
"""
def __init__(self, props=None):
"""
Init routine for class get_from_web.
@param props: myapptools.AppProp instance with proxy information. This has to be a dictionary (see self._set_props).
"""
self.props = props
if props != None:
# install a callback if property 'Proxy' had been changed. it will be called with the new proxy information.
props.InstallPostSetCallback('Proxy', self._props_callback)
def _props_callback(self, proxy):
"""
Routione which is called, if proxy information had been changed. It will set the new proxy information.
@param proxy: dictionary with proxy information
"""
self._set_props(**proxy)
def _set_props(self, use_proxy, host, port, use_user, user, use_passwd, passwd=None):
"""
Routine to set the proxy information.
@param host: host to connect to
@param port: port which is used to connect the proxy
@param user: username for proxy
@param passwd: password to be used for user
"""
proxy_str = None
if not use_proxy:
proxy_support = urllib2.ProxyHandler({})
opener = urllib2.build_opener(proxy_support, urllib2.HTTPHandler)
urllib2.install_opener(opener)
elif not use_user:
proxy_str = "http://%s:%d" % (host, port)
elif not use_passwd or passwd == None:
proxy_str = "http://%s@%s:%d" % (user, host, port)
else:
proxy_str = "http://%s:%s@%s:%d" % (user, passwd, host, port)
if proxy_str != None:
proxy_support = urllib2.ProxyHandler({"http": proxy_str})
opener = urllib2.build_opener(proxy_support, urllib2.HTTPHandler)
urllib2.install_opener(opener)
def get_image(self, url):
"""
Routine to download an image from web.
@param url: url to the image
@return: image (type: python Image)
"""
#try:
f = urllib2.urlopen(url)
im = Image.open(StringIO.StringIO(f.read()))
f.close()
return im
#except ???:
# print "exception in get_from_web().get_image('%s')" % url
# return None
class tile_handler(dict):
"""
Class to handle some tile_source classes and a default tile_source.
"""
TILE_SIZE = 256
class tile_source():
"""
Class to get tile by cache or url. It also stores the downloaded tiles to the cache directory.
"""
TILEPATH = 'osm_tiles'
def __init__(self, gfw, name, url, min_zoom, max_zoom):
"""
Init routine for class tile_source.
@param gfw: instance of get_from_web
@param name: name of tile_type
@param url: base url without zoom or tile information
@param min_zoom: minimum existing zoom level for this tile type
@param max_zoom: maximum existing zoom level for this tile type
"""
self.gfw = gfw
self._name = name
self._url = url
self._zooms = range(min_zoom, max_zoom + 1)
def get_path(self, x, y, zoom_lvl):
"""
Routine to get the tile-path information for a specific tile.
@param x: horizontal tile number
@param y: vertical tile number
@param zoom_lvl: zoom level for the tile
@return: path to tile as string
"""
def intstr(num):
return str(int(num))
return os.path.join(__basepath__, self.TILEPATH, self._name, intstr(zoom_lvl), intstr(x), intstr(y) + '.png')
def get_url(self, x, y, zoom_lvl):
"""
Routine to get the url information for a specific tile.
@param x: horizontal tile number
@param y: vertical tile number
@param zoom_lvl: zoom level for the tile
@return: url to tile as string
"""
def intstr(num):
return str(int(num))
return self._url + '/' + intstr(zoom_lvl) + '/' + intstr(x) + '/' + intstr(y) + '.png'
def get_zooms(self):
"""
Routine to get a list of available zoom levels for this source.
@return: zoom levels as a list (e.g. [0,1,2,3]).
"""
return self._zooms
def get_tile(self, x, y, zoom_lvl, max_age, from_cache):
"""
Routine to get a tile.
@param x: horizontal tile number
@param y: vertical tile number
@param zoom_lvl: zoom level for the tile
@param max_age: maximum age where no www-refresh is needed
@param from_cache: if True the tile is from cache
@return: tile as Image
"""
filename = self.get_path(x, y, zoom_lvl)
url = self.get_url(x, y, zoom_lvl)
if from_cache:
try:
return Image.open(filename)
except:
return None
else:
local_time = calendar.timegm(time.gmtime())
try:
tile_tm = os.path.getmtime(filename)
except:
tile_tm = local_time - max_age - 1
if local_time - tile_tm > max_age: # age depending refresh.
im = self.gfw.get_image(url)
try:
self.save(im, filename)
except:
print "exception in tile_handler().get_tile"
#TODO: exception handling.
pass
return im
else:
return None
def save(self, im, filename):
"""
Routine to save the image to cache (directory).
@param im: image to save (type: python Image)
@param filename: name of the file, which will be created
"""
dirname = os.path.dirname(filename)
if not os.path.exists(dirname):
os.makedirs(dirname)
im.save(filename)
def __init__(self, gfw, props=None):
"""
Init routine for class tile_handler
@param gfw: instance of get_from_web
@param props: myapptools.AppProp instance with tilehandler information. This has to be a dictionary (see self._set_proxy).
"""
dict.__init__(self)
self.gfw = gfw
self._active_tile_source = None
self._max_age = 3600
self._append_tile_source(u'OSM-Mapnik', u'http://tile.openstreetmap.org', 0, 18)
self._append_tile_source(u'OSM-CycleMap', u'http://c.tile.opencyclemap.org/cycle', 0, 18)
try:
# install a callback if property 'Tilehandler' had been changed. it will be called with the new tilehandler information.
props.InstallPostSetCallback('Tilehandler', self._props_callback)
except:
#TODO: exception handling
pass
def _props_callback(self, tilehandler):
"""
Routione which is called, if tilehandler information had been changed. It will set the new tilehandler information.
@param tilehandler: dictionary with tilehandler information
"""
self.set_props(**tilehandler)
def set_props(self, source, max_age):
"""
Routine to set the proxy information.
@param source: source for tiles.
@param max_age: maximum age for a tile till it will be refreshed.
"""
self._set_default(source)
self._max_age = max_age
def _append_tile_source(self, name, url, min_zoom, max_zoom):
"""
Routine to append a tilesource.
@param name: Name for this tilesource
@param url: URL for this tile source (without tile depending information e.g. zoom level, ...)
@param min_zoom: Minimum zoom level for this tilesource
@param max_zoom: Maximum zoom level for this tilesource
"""
self[name] = self.tile_source(self.gfw, name, url, min_zoom, max_zoom)
if self._active_tile_source == None:
self._set_default(name)
def _set_default(self, name):
"""
Routine to set the default tilesorce (by name).
@param name: Name for the default tilesource.
@return: True if name was available, False if not.
"""
if name in self.keys():
self._active_tile_source = name
return True
else:
return False
def get_active_source(self):
"""
Routine to get the Name of the active tile source.
@return: name of the active tile source
"""
return self._active_tile_source
def get_max_age(self):
return self._max_age
def get_choices(self):
"""
Routine to get the names of the possible tile sources.
@return: list of possible tile sources
"""
return self.keys()
def get_zooms(self):
"""
Routine to get a list of available zoom levels for this source.
@return: zoom levels as a list (e.g. [0,1,2,3]).
"""
return self[self._active_tile_source].get_zooms()
def get_url(self, x, y, zoom_lvl):
"""
Routine to get the url information for a specific tile.
@param x: horizontal tile number
@param y: vertical tile number
@param zoom_lvl: zoom level for the tile
@return: url to tile as string
"""
return self[self._active_tile_source].get_url(x, y, zoom_lvl)
def get_tile(self, x, y, zoom_lvl, from_cache):
"""
Routine to get a tile.
@param x: horizontal tile number
@param y: vertical tile number
@param zoom_lvl: zoom level for the tile
@param from_cache: if True the tile is from cache
@return: tile as Image
"""
return self[self._active_tile_source].get_tile(x, y, zoom_lvl, self._max_age, from_cache)
def tile_num(self, coordinate, zoom):
"""
Routine which calculates the needed tile for coordinates.
@param coordinate: geo.coordinate instance with geographic information.
@param zoom: zoom information for the needed tile
@return: return a tuple of two float values (x- and y-tile)
"""
lat_rad = math.radians(coordinate[pylibs.geo.coordinate.LATITUDE])
n = 2.0 ** zoom
xtile = (coordinate[pylibs.geo.coordinate.LONGITUDE] + 180.0) / 360.0 * n
ytile = (1.0 - math.log(math.tan(lat_rad) + (1 / math.cos(lat_rad))) / math.pi) / 2.0 * n
return (xtile, ytile)
def coordinate(self, xtile, ytile, zoom):
"""
Routine which calculates geographic information out of tile information.
@param xtile: number of the tile (x)
@param ytile: number of the tile (y)
@param zoom: zoom level
@return: geo.coordinate instance with the geographic information
"""
n = 2.0 ** zoom
lon_deg = xtile / n * 360.0 - 180.0
lat_rad = math.atan(math.sinh(math.pi * (1 - 2 * ytile / n)))
lat_deg = math.degrees(lat_rad)
return pylibs.geo.coordinate(lon=lon_deg, lat=lat_deg)
class osm_map():
"""
This is an Image including an osm map
"""
def __init__(self, th, callback_refresh):
"""
Init routine for osm_map
@param th: tile_handler class needed to get tiles
@param callback_refresh: function to call, if osm_map had been changed (for refresh view)
callback gets two arguments
* the image or None if no image is available yet
* a description text (of the finished issue)
@param border: additional border around specified area
@param vers: refresh behaviour as described in VER_*
"""
self.th = th
self.callback_refresh = callback_refresh
self._del_map_values()
self.running = False
self.stoprequest = False
def _del_map_values(self):
"""
routine to reset (to None) map definitions (e.g. map_resolution, zoom_lvl, view range
"""
self._image = None
self._map_resolution = None
self._zoom_lvl = None
self._center_coordinate = None
def _set_map_values(self, map_resolution, zoom_lvl, center_coordinate):
def zoom_limitation(th, zoom):
if zoom < min(th.get_zooms()):
return min(th.get_zooms())
if zoom > max(th.get_zooms()):
return max(th.get_zooms())
return zoom
self._image = Image.new('RGB', map_resolution, 'white')
self._map_resolution = map_resolution
self._zoom_lvl = zoom_limitation(self.th, zoom_lvl)
self._center_coordinate = center_coordinate
def disable(self):
self.stop_now()
self.callback_refresh = None
def get_image(self):
return self._image
def get_map_resolution(self):
return self._map_resolution
def get_zoom_lvl(self):
return self._zoom_lvl
def get_center_coordinate(self):
return self._center_coordinate
def _paste_tile(self, tile, xy):
"""
routine to paste a single tile at xy in the map image.
@param tile: tile to paste in the image
@param xy: position to paste the tile. also negative or too large values are
possible to paste just parts of a tile
"""
try:
self._image.paste(tile, xy)
except:
print "exception in osm_map()._paste_tile"
#TODO: exception handling
def create_map_by_res_n_centercoord_n_zoomlvl(self, max_x_res, max_y_res, center_coordinate, zoom_lvl, cache_only=False):
"""
routine to ...
@param max_x_res: maximum x resolution
@param max_y_res: maximum y resolution
@param center_coordinate: center coordinates (object of geo.coordinates)
@param zoom_lvl: zoom_level to use for d
"""
if center_coordinate is not None and zoom_lvl is not None:
self._del_map_values()
#
# needed values for further calculations and map creation
#
self._set_map_values((max_x_res, max_y_res), zoom_lvl, center_coordinate)
self._create_map(cache_only)
def create_map_by_coord_n_zoomlvl(self, tl_coord, br_coord, zoom_lvl, cache_only=False):
"""
@param tl_coord: top left coordinates (object of geo.coordinates)
@param br_coord: bottom right coordinates (object of geo.coordinates)
@param zoom_lvl: zoom_level to use for
"""
center_coordinate = pylibs.geo.area(tl_coord, br_coord).center_pos()
tl_tile = self.th.tile_num(tl_coord, zoom_lvl)
br_tile = self.th.tile_num(br_coord, zoom_lvl)
max_x_res = int((br_tile[0] - tl_tile[0]) * self.th.TILE_SIZE)
max_y_res = int((br_tile[1] - tl_tile[1]) * self.th.TILE_SIZE)
self.create_map_by_res_n_centercoord_n_zoomlvl(max_x_res, max_y_res, center_coordinate, zoom_lvl, cache_only)
def create_map_by_res_n_coord(self, max_x_res, max_y_res, tl_coord, br_coord, cache_only=False):
"""
@param max_x_res: maximum x resolution
@param max_y_res: maximum y resolution
@param tl_coord: top left coordinates (object of geo.coordinates)
@param br_coord: bottom right coordinates (object of geo.coordinates)
coord are not the used coordinated for the map corners, cause the zoom_lvl is quatisised
"""
def coordinates_in_map(max_x, max_y, p1, p2, center, zoom_lvl):
tl = self.get_coord_by_xy(0, 0, (max_x, max_y), center, zoom_lvl)
br = self.get_coord_by_xy(max_x, max_y, (max_x, max_y), center, zoom_lvl)
area = pylibs.geo.area(tl, br)
return area.coordinate_in_area(p1) and area.coordinate_in_area(p2)
center_coordinate = pylibs.geo.area(tl_coord, br_coord).center_pos()
zoom_lvl = max(self.th.get_zooms())
while not coordinates_in_map(max_x_res, max_y_res, tl_coord, br_coord, center_coordinate, zoom_lvl):
zoom_lvl -= 1
self.create_map_by_res_n_centercoord_n_zoomlvl(max_x_res, max_y_res, center_coordinate, zoom_lvl, cache_only)
def stop_now(self):
self.stoprequest = True
while self.running:
pass
self.stoprequest = False
def get_coord_by_xy(self, x, y, map_resolution=None, center_coordinate=None, zoom_lvl=None):
zoom_lvl = zoom_lvl or self._zoom_lvl
tl_tile = self._get_tl_tile_num(map_resolution, center_coordinate, zoom_lvl)
xy_tile = (tl_tile[0] + float(x) / self.th.TILE_SIZE, tl_tile[1] + float(y) / self.th.TILE_SIZE)
return self.th.coordinate(xy_tile[0], xy_tile[1], zoom_lvl)
def get_xy_by_coord(self, coord, map_resolution=None, center_coordinate=None, zoom_lvl=None):
tl_tile = self._get_tl_tile_num(map_resolution, center_coordinate, zoom_lvl)
xy_tile = self.th.tile_num(coord, self._zoom_lvl)
x = int((xy_tile[0] - tl_tile[0]) * self.th.TILE_SIZE)
y = int((xy_tile[1] - tl_tile[1]) * self.th.TILE_SIZE)
return (x, y)
def _get_map_res_tiles(self, map_resolution=None):
"""
returns the map resolution in number of tiles
"""
map_resolution = map_resolution or self._map_resolution
if map_resolution:
return (map_resolution[0] / float(self.th.TILE_SIZE), map_resolution[1] / float(self.th.TILE_SIZE))
else:
return None
def _get_tl_tile_num(self, map_resolution=None, center_coordinate=None, zoom_lvl=None):
map_resolution = map_resolution or self._map_resolution
center_coordinate = center_coordinate or self._center_coordinate
zoom_lvl = zoom_lvl or self._zoom_lvl
#
if (map_resolution and center_coordinate and zoom_lvl):
center_tile_num = self.th.tile_num(center_coordinate, zoom_lvl)
map_resolution_tiles = self._get_map_res_tiles(map_resolution)
topleft_tile_num = (center_tile_num[0] - map_resolution_tiles[0] / 2, center_tile_num[1] - map_resolution_tiles[1] / 2)
return topleft_tile_num
else:
return None
def _get_br_tile_num(self, map_resolution=None, center_coordinate=None, zoom_lvl=None):
topleft_tile_num = self._get_tl_tile_num(map_resolution, center_coordinate, zoom_lvl)
map_resolution_tiles = self._get_map_res_tiles(map_resolution)
bottomright_tile_num = (topleft_tile_num[0] + map_resolution_tiles[0], topleft_tile_num[1] + map_resolution_tiles[1])
return bottomright_tile_num
def _get_xy_offset(self):
tl_tile = self._get_tl_tile_num()
x_offs = -int(tl_tile[0] % 1 * self.th.TILE_SIZE)
y_offs = -int(tl_tile[1] % 1 * self.th.TILE_SIZE)
return (x_offs, y_offs)
def _get_tile_list(self):
tl_tile = self._get_tl_tile_num()
br_tile = self._get_br_tile_num()
tile_list = []
for x in range(int(tl_tile[0]), int(br_tile[0]) + 1):
for y in range(int(tl_tile[1]), int(br_tile[1]) + 1):
tile_list.append((x, y, self._zoom_lvl))
return tile_list
def _create_map(self, cache_only):
"""
@param resoultion: map target resolution
@param xy_offset: offset for top left tile (normally <= 0)
@param zoom_lvl: tile zoom_lvl
@param tile_list: list of tiles [[x1, x2, x3], [y1, y2]]
@param description: description text for callback function
"""
def create_map_by_(xy_offset, tile_list, by_path):
#
# create map from already stored tiles (...by_path)
#
num_tiles = len(tile_list)
x0, y0 = tile_list[0][:2]
num = 0
for x, y, z in tile_list:
num += 1
if self.stoprequest:
break
tile = self.th.get_tile(x, y, z, by_path)
if tile != None:
# paste tile only if tile was available
pos = (xy_offset[0] + (x - x0) * self.th.TILE_SIZE, xy_offset[1] + (y - y0) * self.th.TILE_SIZE)
self._paste_tile(tile, pos)
if not by_path:
desc = "Tile " + self.th.get_url(x, y, z) + " added to map."
prog = float(num) / num_tiles
if self.callback_refresh:
self.callback_refresh(self._image, desc, prog)
self.running = True
#TODO: ggf. Klasse um Uebersetzungen zu ermoeglichen.
create_map_by_(self._get_xy_offset(), self._get_tile_list(), by_path=True)
desc = 'Map creation from cache completeled.'
if self.callback_refresh:
self.callback_refresh(self._image, desc, 1.0)
if not cache_only:
create_map_by_(self._get_xy_offset(), self._get_tile_list(), by_path=False)
desc = 'Map creation completeled.'
if self.callback_refresh:
self.callback_refresh(self._image, desc, 1.0)
self.running = False
def show_map(image, description, progress):
print description, "%5.1f%%" % (progress * 100.)
if image != None:
image.show()
'''
Reference in New Issue View Git Blame Copy Permalink