yumoqing/kivyblocks
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2021-01-16 10:12:16 +08:00
parent 26f7c6c47d
commit 5e0d1c60af
23 changed files with 4067 additions and 1671 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kivyblocks/baseWidget.py
View File

@ -10,6 +10,7 @@ from kivy.uix.label import Label
from kivy.event import EventDispatcher
from kivy.metrics import dp
from kivy.core.window import Window
from kivy.clock import Clock
from kivy.uix.actionbar import ActionBar,ActionView,ActionPrevious,ActionItem,ActionButton
from kivy.uix.actionbar import ActionToggleButton, ActionCheck,ActionSeparator,ActionGroup
@ -213,7 +214,6 @@ class Title6(Text):
class Modal(BGColorBehavior, ModalView):
def __init__(self, auto_open=False, color_level=-1, radius=[], **kw):
kw.update({'auto_dismiss':False})
ModalView.__init__(self, **kw)
BGColorBehavior.__init__(self, color_level=color_level,
radius=radius)

3
kivyblocks/bgcolorbehavior.py
View File

@ -48,6 +48,9 @@ class BGColorBehavior(object):
else:
print('on_bgcolor():self.canvas is None')
def is_selected(self):
return self.bgcolor == self.selected_bgcolor
def selected(self):
if self.bgcolor == self.selected_bgcolor:
return

53
kivyblocks/blocks.py
View File

@ -192,7 +192,10 @@ class Blocks(EventDispatcher):
errback=None,**kw):
if url is None:
errback(None,Exception('url is None'))
if errback:
errback(None,Exception('url is None'))
else:
return None
if url.startswith('file://'):
filename = url[7:]
@ -201,19 +204,14 @@ class Blocks(EventDispatcher):
dic = json.loads(b)
return dic
elif url.startswith('http://') or url.startswith('https://'):
"""
h = HTTPDataHandler(url,method=method,params=params,
files=files)
h.bind(on_success=callback)
h.bind(on_error=errback)
h.handle()
"""
try:
hc = HttpClient()
resp=hc(url,method=method,params=params,files=files)
return resp
except Exception as e:
errback(None,e)
if errback:
return errback(None,e)
return None
else:
config = getConfig()
url = config.uihome + url
@ -312,7 +310,7 @@ class Blocks(EventDispatcher):
if isinstance(v,dict) and v.get('widgettype'):
b = Blocks()
w = b.w_build(v)
if hasattr(widgets,k):
if hasattr(widget,k):
aw = getattr(widget,k)
if isinstance(aw,Layout):
aw.add_widget(w)
@ -370,7 +368,7 @@ class Blocks(EventDispatcher):
return
f = self.buildAction(widget,desc)
if f is None:
Logger.Info('Block: get a null function,%s',str(desc))
Logger.info('Block: get a null function,%s',str(desc))
return
w.bind(**{event:f})
@ -392,8 +390,9 @@ class Blocks(EventDispatcher):
def blocksAction(self,widget,desc, *args):
target = Blocks.getWidgetById(desc.get('target','self'),widget)
if target is None:
Logger.Info('Block: blocksAction():desc(%s) target not found',
Logger.info('Block: blocksAction():desc(%s) target not found',
str(desc))
return
add_mode = desc.get('mode','replace')
opts = desc.get('options').copy()
d = self.getActionData(widget,desc)
@ -418,8 +417,9 @@ class Blocks(EventDispatcher):
def urlwidgetAction(self,widget,desc, *args):
target = Blocks.getWidgetById(desc.get('target','self'),widget)
if target is None:
Logger.Info('Block: urlwidgetAction():desc(%s) target not found',
Logger.info('Block: urlwidgetAction():desc(%s) target not found',
str(desc))
return
add_mode = desc.get('mode','replace')
opts = desc.get('options').copy()
p = opts.get('params',{}).copy()
@ -470,13 +470,14 @@ class Blocks(EventDispatcher):
target = Blocks.getWidgetById(desc.get('target','self'),
from_widget=widget)
if target is None:
Logger.Info('Block: registedfunctionAction():desc(%s) target not found',
Logger.info('Block: registedfunctionAction():desc(%s) target not found',
str(desc))
return
rf = RegisterFunction()
name = desc.get('rfname')
func = rf.get(name)
if func is None:
Logger.Info('Block: desc=%s rfname(%s) not found',
Logger.info('Block: desc=%s rfname(%s) not found',
str(desc), name)
raise Exception('rfname(%s) not found' % name)
@ -488,13 +489,13 @@ class Blocks(EventDispatcher):
def scriptAction(self, widget, desc, *args):
script = desc.get('script')
if not script:
Logger.Info('Block: scriptAction():desc(%s) target not found',
Logger.info('Block: scriptAction():desc(%s) target not found',
str(desc))
return
target = Blocks.getWidgetById(desc.get('target','self'),
from_widget=widget)
if target is None:
Logger.Info('Block: scriptAction():desc(%s) target not found',
Logger.info('Block: scriptAction():desc(%s) target not found',
str(desc))
d = self.getActionData(widget,desc)
ns = {
@ -512,8 +513,9 @@ class Blocks(EventDispatcher):
method = desc.get('method')
target = Blocks.getWidgetById(desc.get('target','self'),widget)
if target is None:
Logger.Info('Block: methodAction():desc(%s) target not found',
Logger.info('Block: methodAction():desc(%s) target not found',
str(desc))
return
if hasattr(target,method):
f = getattr(target, method)
kwargs = desc.get('options',{}).copy()
@ -536,8 +538,6 @@ class Blocks(EventDispatcher):
]
}
"""
name = desc['widgettype']
def doit(desc):
if not isinstance(desc,dict):
Logger.info('Block: desc must be a dict object',
@ -550,6 +550,12 @@ class Blocks(EventDispatcher):
if hasattr(widget,'ready'):
widget.ready()
if not (isinstance(desc, DictObject) or isinstance(desc, dict)):
print('Block: desc must be a dict object',
desc,type(desc))
self.dispatch('on_failed',Exception('miss url'))
return
while desc['widgettype'] == "urlwidget":
opts = desc.get('options',{}).copy()
extend = desc.get('extend')
@ -564,6 +570,13 @@ class Blocks(EventDispatcher):
if opts.get('url'):
del opts['url']
desc = self.getUrlData(url,**opts)
if not (isinstance(desc, DictObject) or \
isinstance(desc, dict)):
print('Block: desc must be a dict object',
desc,type(desc))
self.dispatch('on_failed',Exception('miss url'))
return
if addon:
desc = dictExtend(desc,addon)
doit(desc)

2
kivyblocks/boxViewer.py
View File

@ -75,7 +75,7 @@ class BoxViewer(WidgetReady, BoxLayout):
self.dataloader.bind(on_submit=self.getParams)
self.add_widget(self.viewContainer)
self.register_event_type('on_selected')
self.viewContainer.bind(size=self.resetCols,
self.bind(size=self.resetCols,
pos=self.resetCols)
self.viewContainer.bind(on_scroll_stop=self.on_scroll_stop)
# use_keyboard() will block the python

BIN
kivyblocks/graph/.__init__.py.swp
View File

Binary file not shown.

2995
kivyblocks/graph/__init__.py
View File

File diff suppressed because it is too large Load Diff

27
kivyblocks/mapview/__init__.py Normal file
View File

@ -0,0 +1,27 @@
# coding=utf-8
"""
MapView
=======
MapView is a Kivy widget that display maps.
"""
from kivyblocks.mapview.source import MapSource
from kivyblocks.mapview.types import Bbox, Coordinate
from kivyblocks.mapview.view import (
MapLayer,
MapMarker,
MapMarkerPopup,
MapView,
MarkerMapLayer,
)
__all__ = [
"Coordinate",
"Bbox",
"MapView",
"MapSource",
"MapMarker",
"MapLayer",
"MarkerMapLayer",
"MapMarkerPopup",
]

1
kivyblocks/mapview/_version.py Normal file
View File

@ -0,0 +1 @@
__version__ = "1.0.5"

449
kivyblocks/mapview/clustered_marker_layer.py Normal file
View File

@ -0,0 +1,449 @@
# coding=utf-8
"""
Layer that support point clustering
===================================
"""
from math import atan, exp, floor, log, pi, sin, sqrt
from os.path import dirname, join
from kivy.lang import Builder
from kivy.metrics import dp
from kivy.properties import (
ListProperty,
NumericProperty,
ObjectProperty,
StringProperty,
)
from kivyblocks.mapview.view import MapLayer, MapMarker
Builder.load_string(
"""
<ClusterMapMarker>:
size_hint: None, None
source: root.source
size: list(map(dp, self.texture_size))
allow_stretch: True
Label:
color: root.text_color
pos: root.pos
size: root.size
text: "{}".format(root.num_points)
font_size: dp(18)
"""
)
# longitude/latitude to spherical mercator in [0..1] range
def lngX(lng):
return lng / 360.0 + 0.5
def latY(lat):
if lat == 90:
return 0
if lat == -90:
return 1
s = sin(lat * pi / 180.0)
y = 0.5 - 0.25 * log((1 + s) / (1 - s)) / pi
return min(1, max(0, y))
# spherical mercator to longitude/latitude
def xLng(x):
return (x - 0.5) * 360
def yLat(y):
y2 = (180 - y * 360) * pi / 180
return 360 * atan(exp(y2)) / pi - 90
class KDBush:
"""
kdbush implementation from:
https://github.com/mourner/kdbush/blob/master/src/kdbush.js
"""
def __init__(self, points, node_size=64):
self.points = points
self.node_size = node_size
self.ids = ids = [0] * len(points)
self.coords = coords = [0] * len(points) * 2
for i, point in enumerate(points):
ids[i] = i
coords[2 * i] = point.x
coords[2 * i + 1] = point.y
self._sort(ids, coords, node_size, 0, len(ids) - 1, 0)
def range(self, min_x, min_y, max_x, max_y):
return self._range(
self.ids, self.coords, min_x, min_y, max_x, max_y, self.node_size
)
def within(self, x, y, r):
return self._within(self.ids, self.coords, x, y, r, self.node_size)
def _sort(self, ids, coords, node_size, left, right, depth):
if right - left <= node_size:
return
m = int(floor((left + right) / 2.0))
self._select(ids, coords, m, left, right, depth % 2)
self._sort(ids, coords, node_size, left, m - 1, depth + 1)
self._sort(ids, coords, node_size, m + 1, right, depth + 1)
def _select(self, ids, coords, k, left, right, inc):
swap_item = self._swap_item
while right > left:
if (right - left) > 600:
n = float(right - left + 1)
m = k - left + 1
z = log(n)
s = 0.5 + exp(2 * z / 3.0)
sd = 0.5 * sqrt(z * s * (n - s) / n) * (-1 if (m - n / 2.0) < 0 else 1)
new_left = max(left, int(floor(k - m * s / n + sd)))
new_right = min(right, int(floor(k + (n - m) * s / n + sd)))
self._select(ids, coords, k, new_left, new_right, inc)
t = coords[2 * k + inc]
i = left
j = right
swap_item(ids, coords, left, k)
if coords[2 * right + inc] > t:
swap_item(ids, coords, left, right)
while i < j:
swap_item(ids, coords, i, j)
i += 1
j -= 1
while coords[2 * i + inc] < t:
i += 1
while coords[2 * j + inc] > t:
j -= 1
if coords[2 * left + inc] == t:
swap_item(ids, coords, left, j)
else:
j += 1
swap_item(ids, coords, j, right)
if j <= k:
left = j + 1
if k <= j:
right = j - 1
def _swap_item(self, ids, coords, i, j):
swap = self._swap
swap(ids, i, j)
swap(coords, 2 * i, 2 * j)
swap(coords, 2 * i + 1, 2 * j + 1)
def _swap(self, arr, i, j):
tmp = arr[i]
arr[i] = arr[j]
arr[j] = tmp
def _range(self, ids, coords, min_x, min_y, max_x, max_y, node_size):
stack = [0, len(ids) - 1, 0]
result = []
x = y = 0
while stack:
axis = stack.pop()
right = stack.pop()
left = stack.pop()
if right - left <= node_size:
for i in range(left, right + 1):
x = coords[2 * i]
y = coords[2 * i + 1]
if x >= min_x and x <= max_x and y >= min_y and y <= max_y:
result.append(ids[i])
continue
m = int(floor((left + right) / 2.0))
x = coords[2 * m]
y = coords[2 * m + 1]
if x >= min_x and x <= max_x and y >= min_y and y <= max_y:
result.append(ids[m])
nextAxis = (axis + 1) % 2
if min_x <= x if axis == 0 else min_y <= y:
stack.append(left)
stack.append(m - 1)
stack.append(nextAxis)
if max_x >= x if axis == 0 else max_y >= y:
stack.append(m + 1)
stack.append(right)
stack.append(nextAxis)
return result
def _within(self, ids, coords, qx, qy, r, node_size):
sq_dist = self._sq_dist
stack = [0, len(ids) - 1, 0]
result = []
r2 = r * r
while stack:
axis = stack.pop()
right = stack.pop()
left = stack.pop()
if right - left <= node_size:
for i in range(left, right + 1):
if sq_dist(coords[2 * i], coords[2 * i + 1], qx, qy) <= r2:
result.append(ids[i])
continue
m = int(floor((left + right) / 2.0))
x = coords[2 * m]
y = coords[2 * m + 1]
if sq_dist(x, y, qx, qy) <= r2:
result.append(ids[m])
nextAxis = (axis + 1) % 2
if (qx - r <= x) if axis == 0 else (qy - r <= y):
stack.append(left)
stack.append(m - 1)
stack.append(nextAxis)
if (qx + r >= x) if axis == 0 else (qy + r >= y):
stack.append(m + 1)
stack.append(right)
stack.append(nextAxis)
return result
def _sq_dist(self, ax, ay, bx, by):
dx = ax - bx
dy = ay - by
return dx * dx + dy * dy
class Cluster:
def __init__(self, x, y, num_points, id, props):
self.x = x
self.y = y
self.num_points = num_points
self.zoom = float("inf")
self.id = id
self.props = props
self.parent_id = None
self.widget = None
# preprocess lon/lat
self.lon = xLng(x)
self.lat = yLat(y)
class Marker:
def __init__(self, lon, lat, cls=MapMarker, options=None):
self.lon = lon
self.lat = lat
self.cls = cls
self.options = options
# preprocess x/y from lon/lat
self.x = lngX(lon)
self.y = latY(lat)
# cluster information
self.id = None
self.zoom = float("inf")
self.parent_id = None
self.widget = None
def __repr__(self):
return "<Marker lon={} lat={} source={}>".format(
self.lon, self.lat, self.source
)
class SuperCluster:
"""Port of supercluster from mapbox in pure python
"""
def __init__(self, min_zoom=0, max_zoom=16, radius=40, extent=512, node_size=64):
self.min_zoom = min_zoom
self.max_zoom = max_zoom
self.radius = radius
self.extent = extent
self.node_size = node_size
def load(self, points):
"""Load an array of markers.
Once loaded, the index is immutable.
"""
from time import time
self.trees = {}
self.points = points
for index, point in enumerate(points):
point.id = index
clusters = points
for z in range(self.max_zoom, self.min_zoom - 1, -1):
start = time()
print("build tree", z)
self.trees[z + 1] = KDBush(clusters, self.node_size)
print("kdbush", (time() - start) * 1000)
start = time()
clusters = self._cluster(clusters, z)
print(len(clusters))
print("clustering", (time() - start) * 1000)
self.trees[self.min_zoom] = KDBush(clusters, self.node_size)
def get_clusters(self, bbox, zoom):
"""For the given bbox [westLng, southLat, eastLng, northLat], and
integer zoom, returns an array of clusters and markers
"""
tree = self.trees[self._limit_zoom(zoom)]
ids = tree.range(lngX(bbox[0]), latY(bbox[3]), lngX(bbox[2]), latY(bbox[1]))
clusters = []
for i in range(len(ids)):
c = tree.points[ids[i]]
if isinstance(c, Cluster):
clusters.append(c)
else:
clusters.append(self.points[c.id])
return clusters
def _limit_zoom(self, z):
return max(self.min_zoom, min(self.max_zoom + 1, z))
def _cluster(self, points, zoom):
clusters = []
c_append = clusters.append
trees = self.trees
r = self.radius / float(self.extent * pow(2, zoom))
# loop through each point
for i in range(len(points)):
p = points[i]
# if we've already visited the point at this zoom level, skip it
if p.zoom <= zoom:
continue
p.zoom = zoom
# find all nearby points
tree = trees[zoom + 1]
neighbor_ids = tree.within(p.x, p.y, r)
num_points = 1
if isinstance(p, Cluster):
num_points = p.num_points
wx = p.x * num_points
wy = p.y * num_points
props = None
for j in range(len(neighbor_ids)):
b = tree.points[neighbor_ids[j]]
# filter out neighbors that are too far or already processed
if zoom < b.zoom:
num_points2 = 1
if isinstance(b, Cluster):
num_points2 = b.num_points
# save the zoom (so it doesn't get processed twice)
b.zoom = zoom
# accumulate coordinates for calculating weighted center
wx += b.x * num_points2
wy += b.y * num_points2
num_points += num_points2
b.parent_id = i
if num_points == 1:
c_append(p)
else:
p.parent_id = i
c_append(
Cluster(wx / num_points, wy / num_points, num_points, i, props)
)
return clusters
class ClusterMapMarker(MapMarker):
source = StringProperty(join(dirname(__file__), "icons", "cluster.png"))
cluster = ObjectProperty()
num_points = NumericProperty()
text_color = ListProperty([0.1, 0.1, 0.1, 1])
def on_cluster(self, instance, cluster):
self.num_points = cluster.num_points
def on_touch_down(self, touch):
return False
class ClusteredMarkerLayer(MapLayer):
cluster_cls = ObjectProperty(ClusterMapMarker)
cluster_min_zoom = NumericProperty(0)
cluster_max_zoom = NumericProperty(16)
cluster_radius = NumericProperty("40dp")
cluster_extent = NumericProperty(512)
cluster_node_size = NumericProperty(64)
def __init__(self, **kwargs):
self.cluster = None
self.cluster_markers = []
super().__init__(**kwargs)
def add_marker(self, lon, lat, cls=MapMarker, options=None):
if options is None:
options = {}
marker = Marker(lon, lat, cls, options)
self.cluster_markers.append(marker)
return marker
def remove_marker(self, marker):
self.cluster_markers.remove(marker)
def reposition(self):
if self.cluster is None:
self.build_cluster()
margin = dp(48)
mapview = self.parent
set_marker_position = self.set_marker_position
bbox = mapview.get_bbox(margin)
bbox = (bbox[1], bbox[0], bbox[3], bbox[2])
self.clear_widgets()
for point in self.cluster.get_clusters(bbox, mapview.zoom):
widget = point.widget
if widget is None:
widget = self.create_widget_for(point)
set_marker_position(mapview, widget)
self.add_widget(widget)
def build_cluster(self):
self.cluster = SuperCluster(
min_zoom=self.cluster_min_zoom,
max_zoom=self.cluster_max_zoom,
radius=self.cluster_radius,
extent=self.cluster_extent,
node_size=self.cluster_node_size,
)
self.cluster.load(self.cluster_markers)
def create_widget_for(self, point):
if isinstance(point, Marker):
point.widget = point.cls(lon=point.lon, lat=point.lat, **point.options)
elif isinstance(point, Cluster):
point.widget = self.cluster_cls(lon=point.lon, lat=point.lat, cluster=point)
return point.widget
def set_marker_position(self, mapview, marker):
x, y = mapview.get_window_xy_from(marker.lat, marker.lon, mapview.zoom)
marker.x = int(x - marker.width * marker.anchor_x)
marker.y = int(y - marker.height * marker.anchor_y)

5
kivyblocks/mapview/constants.py Normal file
View File

@ -0,0 +1,5 @@
MIN_LATITUDE = -90.0
MAX_LATITUDE = 90.0
MIN_LONGITUDE = -180.0
MAX_LONGITUDE = 180.0
CACHE_DIR = "cache"

123
kivyblocks/mapview/downloader.py Normal file
View File

@ -0,0 +1,123 @@
# coding=utf-8
__all__ = ["Downloader"]
import logging
import traceback
from concurrent.futures import ThreadPoolExecutor, TimeoutError, as_completed
from os import environ, makedirs
from os.path import exists, join
from random import choice
from time import time
import requests
from kivy.clock import Clock
from kivy.logger import LOG_LEVELS, Logger
from kivyblocks.mapview.constants import CACHE_DIR
if "MAPVIEW_DEBUG_DOWNLOADER" in environ:
Logger.setLevel(LOG_LEVELS['debug'])
# user agent is needed because since may 2019 OSM gives me a 429 or 403 server error
# I tried it with a simpler one (just Mozilla/5.0) this also gets rejected
USER_AGENT = 'Kivy-garden.mapview'
class Downloader:
_instance = None
MAX_WORKERS = 5
CAP_TIME = 0.064 # 15 FPS
@staticmethod
def instance(cache_dir=None):
if Downloader._instance is None:
if not cache_dir:
cache_dir = CACHE_DIR
Downloader._instance = Downloader(cache_dir=cache_dir)
return Downloader._instance
def __init__(self, max_workers=None, cap_time=None, **kwargs):
self.cache_dir = kwargs.get('cache_dir', CACHE_DIR)
if max_workers is None:
max_workers = Downloader.MAX_WORKERS
if cap_time is None:
cap_time = Downloader.CAP_TIME
self.is_paused = False
self.cap_time = cap_time
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self._futures = []
Clock.schedule_interval(self._check_executor, 1 / 60.0)
if not exists(self.cache_dir):
makedirs(self.cache_dir)
def submit(self, f, *args, **kwargs):
future = self.executor.submit(f, *args, **kwargs)
self._futures.append(future)
def download_tile(self, tile):
Logger.debug(
"Downloader: queue(tile) zoom={} x={} y={}".format(
tile.zoom, tile.tile_x, tile.tile_y
)
)
future = self.executor.submit(self._load_tile, tile)
self._futures.append(future)
def download(self, url, callback, **kwargs):
Logger.debug("Downloader: queue(url) {}".format(url))
future = self.executor.submit(self._download_url, url, callback, kwargs)
self._futures.append(future)
def _download_url(self, url, callback, kwargs):
Logger.debug("Downloader: download(url) {}".format(url))
response = requests.get(url, **kwargs)
response.raise_for_status()
return callback, (url, response)
def _load_tile(self, tile):
if tile.state == "done":
return
cache_fn = tile.cache_fn
if exists(cache_fn):
Logger.debug("Downloader: use cache {}".format(cache_fn))
return tile.set_source, (cache_fn,)
tile_y = tile.map_source.get_row_count(tile.zoom) - tile.tile_y - 1
uri = tile.map_source.url.format(
z=tile.zoom, x=tile.tile_x, y=tile_y, s=choice(tile.map_source.subdomains)
)
Logger.debug("Downloader: download(tile) {}".format(uri))
response = requests.get(uri, headers={'User-agent': USER_AGENT}, timeout=5)
try:
response.raise_for_status()
data = response.content
with open(cache_fn, "wb") as fd:
fd.write(data)
Logger.debug("Downloaded {} bytes: {}".format(len(data), uri))
return tile.set_source, (cache_fn,)
except Exception as e:
print("Downloader error: {!r}".format(e))
def _check_executor(self, dt):
start = time()
try:
for future in as_completed(self._futures[:], 0):
self._futures.remove(future)
try:
result = future.result()
except Exception:
traceback.print_exc()
# make an error tile?
continue
if result is None:
continue
callback, args = result
callback(*args)
# capped executor in time, in order to prevent too much
# slowiness.
# seems to works quite great with big zoom-in/out
if time() - start > self.cap_time:
break
except TimeoutError:
pass

381
kivyblocks/mapview/geojson.py Normal file
View File

@ -0,0 +1,381 @@
# coding=utf-8
"""
Geojson layer
=============
.. note::
Currently experimental and a work in progress, not fully optimized.
Supports:
- html color in properties
- polygon geometry are cached and not redrawed when the parent mapview changes
- linestring are redrawed everymove, it's ugly and slow.
- marker are NOT supported
"""
__all__ = ["GeoJsonMapLayer"]
import json
from kivy.graphics import (
Canvas,
Color,
Line,
MatrixInstruction,
Mesh,
PopMatrix,
PushMatrix,
Scale,
Translate,
)
from kivy.graphics.tesselator import TYPE_POLYGONS, WINDING_ODD, Tesselator
from kivy.metrics import dp
from kivy.properties import ObjectProperty, StringProperty
from kivy.utils import get_color_from_hex
from kivyblocks.mapview.constants import CACHE_DIR
from kivyblocks.mapview.downloader import Downloader
from kivyblocks.mapview.view import MapLayer
COLORS = {
'aliceblue': '#f0f8ff',
'antiquewhite': '#faebd7',
'aqua': '#00ffff',
'aquamarine': '#7fffd4',
'azure': '#f0ffff',
'beige': '#f5f5dc',
'bisque': '#ffe4c4',
'black': '#000000',
'blanchedalmond': '#ffebcd',
'blue': '#0000ff',
'blueviolet': '#8a2be2',
'brown': '#a52a2a',
'burlywood': '#deb887',
'cadetblue': '#5f9ea0',
'chartreuse': '#7fff00',
'chocolate': '#d2691e',
'coral': '#ff7f50',
'cornflowerblue': '#6495ed',
'cornsilk': '#fff8dc',
'crimson': '#dc143c',
'cyan': '#00ffff',
'darkblue': '#00008b',
'darkcyan': '#008b8b',
'darkgoldenrod': '#b8860b',
'darkgray': '#a9a9a9',
'darkgrey': '#a9a9a9',
'darkgreen': '#006400',
'darkkhaki': '#bdb76b',
'darkmagenta': '#8b008b',
'darkolivegreen': '#556b2f',
'darkorange': '#ff8c00',
'darkorchid': '#9932cc',
'darkred': '#8b0000',
'darksalmon': '#e9967a',
'darkseagreen': '#8fbc8f',
'darkslateblue': '#483d8b',
'darkslategray': '#2f4f4f',
'darkslategrey': '#2f4f4f',
'darkturquoise': '#00ced1',
'darkviolet': '#9400d3',
'deeppink': '#ff1493',
'deepskyblue': '#00bfff',
'dimgray': '#696969',
'dimgrey': '#696969',
'dodgerblue': '#1e90ff',
'firebrick': '#b22222',
'floralwhite': '#fffaf0',
'forestgreen': '#228b22',
'fuchsia': '#ff00ff',
'gainsboro': '#dcdcdc',
'ghostwhite': '#f8f8ff',
'gold': '#ffd700',
'goldenrod': '#daa520',
'gray': '#808080',
'grey': '#808080',
'green': '#008000',
'greenyellow': '#adff2f',
'honeydew': '#f0fff0',
'hotpink': '#ff69b4',
'indianred': '#cd5c5c',
'indigo': '#4b0082',
'ivory': '#fffff0',
'khaki': '#f0e68c',
'lavender': '#e6e6fa',
'lavenderblush': '#fff0f5',
'lawngreen': '#7cfc00',
'lemonchiffon': '#fffacd',
'lightblue': '#add8e6',
'lightcoral': '#f08080',
'lightcyan': '#e0ffff',
'lightgoldenrodyellow': '#fafad2',
'lightgray': '#d3d3d3',
'lightgrey': '#d3d3d3',
'lightgreen': '#90ee90',
'lightpink': '#ffb6c1',
'lightsalmon': '#ffa07a',
'lightseagreen': '#20b2aa',
'lightskyblue': '#87cefa',
'lightslategray': '#778899',
'lightslategrey': '#778899',
'lightsteelblue': '#b0c4de',
'lightyellow': '#ffffe0',
'lime': '#00ff00',
'limegreen': '#32cd32',
'linen': '#faf0e6',
'magenta': '#ff00ff',
'maroon': '#800000',
'mediumaquamarine': '#66cdaa',
'mediumblue': '#0000cd',
'mediumorchid': '#ba55d3',
'mediumpurple': '#9370d8',
'mediumseagreen': '#3cb371',
'mediumslateblue': '#7b68ee',
'mediumspringgreen': '#00fa9a',
'mediumturquoise': '#48d1cc',
'mediumvioletred': '#c71585',
'midnightblue': '#191970',
'mintcream': '#f5fffa',
'mistyrose': '#ffe4e1',
'moccasin': '#ffe4b5',
'navajowhite': '#ffdead',
'navy': '#000080',
'oldlace': '#fdf5e6',
'olive': '#808000',
'olivedrab': '#6b8e23',
'orange': '#ffa500',
'orangered': '#ff4500',
'orchid': '#da70d6',
'palegoldenrod': '#eee8aa',
'palegreen': '#98fb98',
'paleturquoise': '#afeeee',
'palevioletred': '#d87093',
'papayawhip': '#ffefd5',
'peachpuff': '#ffdab9',
'peru': '#cd853f',
'pink': '#ffc0cb',
'plum': '#dda0dd',
'powderblue': '#b0e0e6',
'purple': '#800080',
'red': '#ff0000',
'rosybrown': '#bc8f8f',
'royalblue': '#4169e1',
'saddlebrown': '#8b4513',
'salmon': '#fa8072',
'sandybrown': '#f4a460',
'seagreen': '#2e8b57',
'seashell': '#fff5ee',
'sienna': '#a0522d',
'silver': '#c0c0c0',
'skyblue': '#87ceeb',
'slateblue': '#6a5acd',
'slategray': '#708090',
'slategrey': '#708090',
'snow': '#fffafa',
'springgreen': '#00ff7f',
'steelblue': '#4682b4',
'tan': '#d2b48c',
'teal': '#008080',
'thistle': '#d8bfd8',
'tomato': '#ff6347',
'turquoise': '#40e0d0',
'violet': '#ee82ee',
'wheat': '#f5deb3',
'white': '#ffffff',
'whitesmoke': '#f5f5f5',
'yellow': '#ffff00',
'yellowgreen': '#9acd32',
}
def flatten(lst):
return [item for sublist in lst for item in sublist]
class GeoJsonMapLayer(MapLayer):
source = StringProperty()
geojson = ObjectProperty()
cache_dir = StringProperty(CACHE_DIR)
def __init__(self, **kwargs):
self.first_time = True
self.initial_zoom = None
super().__init__(**kwargs)
with self.canvas:
self.canvas_polygon = Canvas()
self.canvas_line = Canvas()
with self.canvas_polygon.before:
PushMatrix()
self.g_matrix = MatrixInstruction()
self.g_scale = Scale()
self.g_translate = Translate()
with self.canvas_polygon:
self.g_canvas_polygon = Canvas()
with self.canvas_polygon.after:
PopMatrix()
def reposition(self):
vx, vy = self.parent.delta_x, self.parent.delta_y
pzoom = self.parent.zoom
zoom = self.initial_zoom
if zoom is None:
self.initial_zoom = zoom = pzoom
if zoom != pzoom:
diff = 2 ** (pzoom - zoom)
vx /= diff
vy /= diff
self.g_scale.x = self.g_scale.y = diff
else:
self.g_scale.x = self.g_scale.y = 1.0
self.g_translate.xy = vx, vy
self.g_matrix.matrix = self.parent._scatter.transform
if self.geojson:
update = not self.first_time
self.on_geojson(self, self.geojson, update=update)
self.first_time = False
def traverse_feature(self, func, part=None):
"""Traverse the whole geojson and call the func with every element
found.
"""
if part is None:
part = self.geojson
if not part:
return
tp = part["type"]
if tp == "FeatureCollection":
for feature in part["features"]:
func(feature)
elif tp == "Feature":
func(part)
@property
def bounds(self):
# return the min lon, max lon, min lat, max lat
bounds = [float("inf"), float("-inf"), float("inf"), float("-inf")]
def _submit_coordinate(coord):
lon, lat = coord
bounds[0] = min(bounds[0], lon)
bounds[1] = max(bounds[1], lon)
bounds[2] = min(bounds[2], lat)
bounds[3] = max(bounds[3], lat)
def _get_bounds(feature):
geometry = feature["geometry"]
tp = geometry["type"]
if tp == "Point":
_submit_coordinate(geometry["coordinates"])
elif tp == "Polygon":
for coordinate in geometry["coordinates"][0]:
_submit_coordinate(coordinate)
elif tp == "MultiPolygon":
for polygon in geometry["coordinates"]:
for coordinate in polygon[0]:
_submit_coordinate(coordinate)
self.traverse_feature(_get_bounds)
return bounds
@property
def center(self):
min_lon, max_lon, min_lat, max_lat = self.bounds
cx = (max_lon - min_lon) / 2.0
cy = (max_lat - min_lat) / 2.0
return min_lon + cx, min_lat + cy
def on_geojson(self, instance, geojson, update=False):
if self.parent is None:
return
if not update:
self.g_canvas_polygon.clear()
self._geojson_part(geojson, geotype="Polygon")
self.canvas_line.clear()
self._geojson_part(geojson, geotype="LineString")
def on_source(self, instance, value):
if value.startswith(("http://", "https://")):
Downloader.instance(cache_dir=self.cache_dir).download(
value, self._load_geojson_url
)
else:
with open(value, "rb") as fd:
geojson = json.load(fd)
self.geojson = geojson
def _load_geojson_url(self, url, response):
self.geojson = response.json()
def _geojson_part(self, part, geotype=None):
tp = part["type"]
if tp == "FeatureCollection":
for feature in part["features"]:
if geotype and feature["geometry"]["type"] != geotype:
continue
self._geojson_part_f(feature)
elif tp == "Feature":
if geotype and part["geometry"]["type"] == geotype:
self._geojson_part_f(part)
else:
# unhandled geojson part
pass
def _geojson_part_f(self, feature):
properties = feature["properties"]
geometry = feature["geometry"]
graphics = self._geojson_part_geometry(geometry, properties)
for g in graphics:
tp = geometry["type"]
if tp == "Polygon":
self.g_canvas_polygon.add(g)
else:
self.canvas_line.add(g)
def _geojson_part_geometry(self, geometry, properties):
tp = geometry["type"]
graphics = []
if tp == "Polygon":
tess = Tesselator()
for c in geometry["coordinates"]:
xy = list(self._lonlat_to_xy(c))
xy = flatten(xy)
tess.add_contour(xy)
tess.tesselate(WINDING_ODD, TYPE_POLYGONS)
color = self._get_color_from(properties.get("color", "FF000088"))
graphics.append(Color(*color))
for vertices, indices in tess.meshes:
graphics.append(
Mesh(vertices=vertices, indices=indices, mode="triangle_fan")
)
elif tp == "LineString":
stroke = get_color_from_hex(properties.get("stroke", "#ffffff"))
stroke_width = dp(properties.get("stroke-width"))
xy = list(self._lonlat_to_xy(geometry["coordinates"]))
xy = flatten(xy)
graphics.append(Color(*stroke))
graphics.append(Line(points=xy, width=stroke_width))
return graphics
def _lonlat_to_xy(self, lonlats):
view = self.parent
zoom = view.zoom
for lon, lat in lonlats:
p = view.get_window_xy_from(lat, lon, zoom)
p = p[0] - self.parent.delta_x, p[1] - self.parent.delta_y
p = self.parent._scatter.to_local(*p)
yield p
def _get_color_from(self, value):
color = COLORS.get(value.lower(), value)
color = get_color_from_hex(color)
return color

BIN
kivyblocks/mapview/icons/cluster.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 600 B

BIN
kivyblocks/mapview/icons/marker.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 4.1 KiB

121
kivyblocks/mapview/mbtsource.py Normal file
View File

@ -0,0 +1,121 @@
# coding=utf-8
"""
MBTiles provider for MapView
============================
This provider is based on .mbfiles from MapBox.
See: http://mbtiles.org/
"""
__all__ = ["MBTilesMapSource"]
import io
import sqlite3
import threading
from kivy.core.image import Image as CoreImage
from kivy.core.image import ImageLoader
from kivyblocks.mapview.downloader import Downloader
from kivyblocks.mapview.source import MapSource
class MBTilesMapSource(MapSource):
def __init__(self, filename, **kwargs):
super().__init__(**kwargs)
self.filename = filename
self.db = sqlite3.connect(filename)
# read metadata
c = self.db.cursor()
metadata = dict(c.execute("SELECT * FROM metadata"))
if metadata["format"] == "pbf":
raise ValueError("Only raster maps are supported, not vector maps.")
self.min_zoom = int(metadata["minzoom"])
self.max_zoom = int(metadata["maxzoom"])
self.attribution = metadata.get("attribution", "")
self.bounds = bounds = None
cx = cy = 0.0
cz = 5
if "bounds" in metadata:
self.bounds = bounds = map(float, metadata["bounds"].split(","))
if "center" in metadata:
cx, cy, cz = map(float, metadata["center"].split(","))
elif self.bounds:
cx = (bounds[2] + bounds[0]) / 2.0
cy = (bounds[3] + bounds[1]) / 2.0
cz = self.min_zoom
self.default_lon = cx
self.default_lat = cy
self.default_zoom = int(cz)
self.projection = metadata.get("projection", "")
self.is_xy = self.projection == "xy"
def fill_tile(self, tile):
if tile.state == "done":
return
Downloader.instance(self.cache_dir).submit(self._load_tile, tile)
def _load_tile(self, tile):
# global db context cannot be shared across threads.
ctx = threading.local()
if not hasattr(ctx, "db"):
ctx.db = sqlite3.connect(self.filename)
# get the right tile
c = ctx.db.cursor()
c.execute(
(
"SELECT tile_data FROM tiles WHERE "
"zoom_level=? AND tile_column=? AND tile_row=?"
),
(tile.zoom, tile.tile_x, tile.tile_y),
)
row = c.fetchone()
if not row:
tile.state = "done"
return
# no-file loading
try:
data = io.BytesIO(row[0])
except Exception:
# android issue, "buffer" does not have the buffer interface
# ie row[0] buffer is not compatible with BytesIO on Android??
data = io.BytesIO(bytes(row[0]))
im = CoreImage(
data,
ext='png',
filename="{}.{}.{}.png".format(tile.zoom, tile.tile_x, tile.tile_y),
)
if im is None:
tile.state = "done"
return
return self._load_tile_done, (tile, im,)
def _load_tile_done(self, tile, im):
tile.texture = im.texture
tile.state = "need-animation"
def get_x(self, zoom, lon):
if self.is_xy:
return lon
return super().get_x(zoom, lon)
def get_y(self, zoom, lat):
if self.is_xy:
return lat
return super().get_y(zoom, lat)
def get_lon(self, zoom, x):
if self.is_xy:
return x
return super().get_lon(zoom, x)
def get_lat(self, zoom, y):
if self.is_xy:
return y
return super().get_lat(zoom, y)

212
kivyblocks/mapview/source.py Normal file
View File

@ -0,0 +1,212 @@
# coding=utf-8
__all__ = ["MapSource"]
import hashlib
from math import atan, ceil, cos, exp, log, pi, tan
from kivy.metrics import dp
from kivyblocks.mapview.constants import (
CACHE_DIR,
MAX_LATITUDE,
MAX_LONGITUDE,
MIN_LATITUDE,
MIN_LONGITUDE,
)
from kivyblocks.mapview.downloader import Downloader
from kivyblocks.mapview.utils import clamp
class MapSource:
"""Base class for implementing a map source / provider
"""
attribution_osm = 'Maps & Data © [i][ref=http://www.osm.org/copyright]OpenStreetMap contributors[/ref][/i]'
attribution_thunderforest = 'Maps © [i][ref=http://www.thunderforest.com]Thunderforest[/ref][/i], Data © [i][ref=http://www.osm.org/copyright]OpenStreetMap contributors[/ref][/i]'
# list of available providers
# cache_key: (is_overlay, minzoom, maxzoom, url, attribution)
providers = {
"osm": (
0,
0,
19,
"http://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png",
attribution_osm,
),
"osm-hot": (
0,
0,
19,
"http://{s}.tile.openstreetmap.fr/hot/{z}/{x}/{y}.png",
"",
),
"osm-de": (
0,
0,
18,
"http://{s}.tile.openstreetmap.de/tiles/osmde/{z}/{x}/{y}.png",
"Tiles @ OSM DE",
),
"osm-fr": (
0,
0,
20,
"http://{s}.tile.openstreetmap.fr/osmfr/{z}/{x}/{y}.png",
"Tiles @ OSM France",
),
"cyclemap": (
0,
0,
17,
"http://{s}.tile.opencyclemap.org/cycle/{z}/{x}/{y}.png",
"Tiles @ Andy Allan",
),
"thunderforest-cycle": (
0,
0,
19,
"http://{s}.tile.thunderforest.com/cycle/{z}/{x}/{y}.png",
attribution_thunderforest,
),
"thunderforest-transport": (
0,
0,
19,
"http://{s}.tile.thunderforest.com/transport/{z}/{x}/{y}.png",
attribution_thunderforest,
),
"thunderforest-landscape": (
0,
0,
19,
"http://{s}.tile.thunderforest.com/landscape/{z}/{x}/{y}.png",
attribution_thunderforest,
),
"thunderforest-outdoors": (
0,
0,
19,
"http://{s}.tile.thunderforest.com/outdoors/{z}/{x}/{y}.png",
attribution_thunderforest,
),
# no longer available
# "mapquest-osm": (0, 0, 19, "http://otile{s}.mqcdn.com/tiles/1.0.0/map/{z}/{x}/{y}.jpeg", "Tiles Courtesy of Mapquest", {"subdomains": "1234", "image_ext": "jpeg"}),
# "mapquest-aerial": (0, 0, 19, "http://oatile{s}.mqcdn.com/tiles/1.0.0/sat/{z}/{x}/{y}.jpeg", "Tiles Courtesy of Mapquest", {"subdomains": "1234", "image_ext": "jpeg"}),
# more to add with
# https://github.com/leaflet-extras/leaflet-providers/blob/master/leaflet-providers.js
# not working ?
# "openseamap": (0, 0, 19, "http://tiles.openseamap.org/seamark/{z}/{x}/{y}.png",
# "Map data @ OpenSeaMap contributors"),
}
def __init__(
self,
url="http://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png",
cache_key=None,
min_zoom=0,
max_zoom=19,
tile_size=256,
image_ext="png",
attribution="© OpenStreetMap contributors",
subdomains="abc",
**kwargs
):
if cache_key is None:
# possible cache hit, but very unlikely
cache_key = hashlib.sha224(url.encode("utf8")).hexdigest()[:10]
self.url = url
self.cache_key = cache_key
self.min_zoom = min_zoom
self.max_zoom = max_zoom
self.tile_size = tile_size
self.image_ext = image_ext
self.attribution = attribution
self.subdomains = subdomains
self.cache_fmt = "{cache_key}_{zoom}_{tile_x}_{tile_y}.{image_ext}"
self.dp_tile_size = min(dp(self.tile_size), self.tile_size * 2)
self.default_lat = self.default_lon = self.default_zoom = None
self.bounds = None
self.cache_dir = kwargs.get('cache_dir', CACHE_DIR)
@staticmethod
def from_provider(key, **kwargs):
provider = MapSource.providers[key]
cache_dir = kwargs.get('cache_dir', CACHE_DIR)
options = {}
is_overlay, min_zoom, max_zoom, url, attribution = provider[:5]
if len(provider) > 5:
options = provider[5]
return MapSource(
cache_key=key,
min_zoom=min_zoom,
max_zoom=max_zoom,
url=url,
cache_dir=cache_dir,
attribution=attribution,
**options
)
def get_x(self, zoom, lon):
"""Get the x position on the map using this map source's projection
(0, 0) is located at the top left.
"""
lon = clamp(lon, MIN_LONGITUDE, MAX_LONGITUDE)
return ((lon + 180.0) / 360.0 * pow(2.0, zoom)) * self.dp_tile_size
def get_y(self, zoom, lat):
"""Get the y position on the map using this map source's projection
(0, 0) is located at the top left.
"""
lat = clamp(-lat, MIN_LATITUDE, MAX_LATITUDE)
lat = lat * pi / 180.0
return (
(1.0 - log(tan(lat) + 1.0 / cos(lat)) / pi) / 2.0 * pow(2.0, zoom)
) * self.dp_tile_size
def get_lon(self, zoom, x):
"""Get the longitude to the x position in the map source's projection
"""
dx = x / float(self.dp_tile_size)
lon = dx / pow(2.0, zoom) * 360.0 - 180.0
return clamp(lon, MIN_LONGITUDE, MAX_LONGITUDE)
def get_lat(self, zoom, y):
"""Get the latitude to the y position in the map source's projection
"""
dy = y / float(self.dp_tile_size)
n = pi - 2 * pi * dy / pow(2.0, zoom)
lat = -180.0 / pi * atan(0.5 * (exp(n) - exp(-n)))
return clamp(lat, MIN_LATITUDE, MAX_LATITUDE)
def get_row_count(self, zoom):
"""Get the number of tiles in a row at this zoom level
"""
if zoom == 0:
return 1
return 2 << (zoom - 1)
def get_col_count(self, zoom):
"""Get the number of tiles in a col at this zoom level
"""
if zoom == 0:
return 1
return 2 << (zoom - 1)
def get_min_zoom(self):
"""Return the minimum zoom of this source
"""
return self.min_zoom
def get_max_zoom(self):
"""Return the maximum zoom of this source
"""
return self.max_zoom
def fill_tile(self, tile):
"""Add this tile to load within the downloader
"""
if tile.state == "done":
return
Downloader.instance(cache_dir=self.cache_dir).download_tile(tile)

29
kivyblocks/mapview/types.py Normal file
View File

@ -0,0 +1,29 @@
# coding=utf-8
__all__ = ["Coordinate", "Bbox"]
from collections import namedtuple
Coordinate = namedtuple("Coordinate", ["lat", "lon"])
class Bbox(tuple):
def collide(self, *args):
if isinstance(args[0], Coordinate):
coord = args[0]
lat = coord.lat
lon = coord.lon
else:
lat, lon = args
lat1, lon1, lat2, lon2 = self[:]
if lat1 < lat2:
in_lat = lat1 <= lat <= lat2
else:
in_lat = lat2 <= lat <= lat2
if lon1 < lon2:
in_lon = lon1 <= lon <= lon2
else:
in_lon = lon2 <= lon <= lon2
return in_lat and in_lon

50
kivyblocks/mapview/utils.py Normal file
View File

@ -0,0 +1,50 @@
# coding=utf-8
__all__ = ["clamp", "haversine", "get_zoom_for_radius"]
from math import asin, cos, pi, radians, sin, sqrt
from kivy.core.window import Window
from kivy.metrics import dp
def clamp(x, minimum, maximum):
return max(minimum, min(x, maximum))
def haversine(lon1, lat1, lon2, lat2):
"""
Calculate the great circle distance between two points
on the earth (specified in decimal degrees)
Taken from: http://stackoverflow.com/questions/4913349/haversine-formula-in-python-bearing-and-distance-between-two-gps-points
"""
# convert decimal degrees to radians
lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])
# haversine formula
dlon = lon2 - lon1
dlat = lat2 - lat1
a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
c = 2 * asin(sqrt(a))
km = 6367 * c
return km
def get_zoom_for_radius(radius_km, lat=None, tile_size=256.0):
"""See: https://wiki.openstreetmap.org/wiki/Zoom_levels"""
radius = radius_km * 1000.0
if lat is None:
lat = 0.0 # Do not compensate for the latitude
# Calculate the equatorial circumference based on the WGS-84 radius
earth_circumference = 2.0 * pi * 6378137.0 * cos(lat * pi / 180.0)
# Check how many tiles that are currently in view
nr_tiles_shown = min(Window.size) / dp(tile_size)
# Keep zooming in until we find a zoom level where the circle can fit inside the screen
zoom = 1
while earth_circumference / (2 << (zoom - 1)) * nr_tiles_shown > 2 * radius:
zoom += 1
return zoom - 1 # Go one zoom level back

999
kivyblocks/mapview/view.py Normal file
View File

@ -0,0 +1,999 @@
# coding=utf-8
__all__ = ["MapView", "MapMarker", "MapMarkerPopup", "MapLayer", "MarkerMapLayer"]
import webbrowser
from itertools import takewhile
from math import ceil
from os.path import dirname, join
from kivy.clock import Clock
from kivy.compat import string_types
from kivy.graphics import Canvas, Color, Rectangle
from kivy.graphics.transformation import Matrix
from kivy.lang import Builder
from kivy.metrics import dp
from kivy.properties import (
AliasProperty,
BooleanProperty,
ListProperty,
NumericProperty,
ObjectProperty,
StringProperty,
)
from kivy.uix.behaviors import ButtonBehavior
from kivy.uix.image import Image
from kivy.uix.label import Label
from kivy.uix.scatter import Scatter
from kivy.uix.widget import Widget
from kivyblocks.mapview import Bbox, Coordinate
from kivyblocks.mapview.constants import (
CACHE_DIR,
MAX_LATITUDE,
MAX_LONGITUDE,
MIN_LATITUDE,
MIN_LONGITUDE,
)
from kivyblocks.mapview.source import MapSource
from kivyblocks.mapview.utils import clamp
Builder.load_string(
"""
<MapMarker>:
size_hint: None, None
source: root.source
size: list(map(dp, self.texture_size))
allow_stretch: True
<MapView>:
canvas.before:
StencilPush
Rectangle:
pos: self.pos
size: self.size
StencilUse
Color:
rgba: self.background_color
Rectangle:
pos: self.pos
size: self.size
canvas.after:
StencilUnUse
Rectangle:
pos: self.pos
size: self.size
StencilPop
ClickableLabel:
text: root.map_source.attribution if hasattr(root.map_source, "attribution") else ""
size_hint: None, None
size: self.texture_size[0] + sp(8), self.texture_size[1] + sp(4)
font_size: "10sp"
right: [root.right, self.center][0]
color: 0, 0, 0, 1
markup: True
canvas.before:
Color:
rgba: .8, .8, .8, .8
Rectangle:
pos: self.pos
size: self.size
<MapViewScatter>:
auto_bring_to_front: False
do_rotation: False
scale_min: 0.2
scale_max: 3.
<MapMarkerPopup>:
RelativeLayout:
id: placeholder
y: root.top
center_x: root.center_x
size: root.popup_size
"""
)
class ClickableLabel(Label):
def on_ref_press(self, *args):
webbrowser.open(str(args[0]), new=2)
class Tile(Rectangle):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cache_dir = kwargs.get('cache_dir', CACHE_DIR)
@property
def cache_fn(self):
map_source = self.map_source
fn = map_source.cache_fmt.format(
image_ext=map_source.image_ext,
cache_key=map_source.cache_key,
**self.__dict__
)
return join(self.cache_dir, fn)
def set_source(self, cache_fn):
self.source = cache_fn
self.state = "need-animation"
class MapMarker(ButtonBehavior, Image):
"""A marker on a map, that must be used on a :class:`MapMarker`
"""
anchor_x = NumericProperty(0.5)
"""Anchor of the marker on the X axis. Defaults to 0.5, mean the anchor will
be at the X center of the image.
"""
anchor_y = NumericProperty(0)
"""Anchor of the marker on the Y axis. Defaults to 0, mean the anchor will
be at the Y bottom of the image.
"""
lat = NumericProperty(0)
"""Latitude of the marker
"""
lon = NumericProperty(0)
"""Longitude of the marker
"""
source = StringProperty(join(dirname(__file__), "icons", "marker.png"))
"""Source of the marker, defaults to our own marker.png
"""
# (internal) reference to its layer
_layer = None
def detach(self):
if self._layer:
self._layer.remove_widget(self)
self._layer = None
class MapMarkerPopup(MapMarker):
is_open = BooleanProperty(False)
placeholder = ObjectProperty(None)
popup_size = ListProperty([100, 100])
def add_widget(self, widget):
if not self.placeholder:
self.placeholder = widget
if self.is_open:
super().add_widget(self.placeholder)
else:
self.placeholder.add_widget(widget)
def remove_widget(self, widget):
if widget is not self.placeholder:
self.placeholder.remove_widget(widget)
else:
super().remove_widget(widget)
def on_is_open(self, *args):
self.refresh_open_status()
def on_release(self, *args):
self.is_open = not self.is_open
def refresh_open_status(self):
if not self.is_open and self.placeholder.parent:
super().remove_widget(self.placeholder)
elif self.is_open and not self.placeholder.parent:
super().add_widget(self.placeholder)
class MapLayer(Widget):
"""A map layer, that is repositionned everytime the :class:`MapView` is
moved.
"""
viewport_x = NumericProperty(0)
viewport_y = NumericProperty(0)
def reposition(self):
"""Function called when :class:`MapView` is moved. You must recalculate
the position of your children.
"""
pass
def unload(self):
"""Called when the view want to completly unload the layer.
"""
pass
class MarkerMapLayer(MapLayer):
"""A map layer for :class:`MapMarker`
"""
order_marker_by_latitude = BooleanProperty(True)
def __init__(self, **kwargs):
self.markers = []
super().__init__(**kwargs)
def insert_marker(self, marker, **kwargs):
if self.order_marker_by_latitude:
before = list(
takewhile(lambda i_m: i_m[1].lat < marker.lat, enumerate(self.children))
)
if before:
kwargs['index'] = before[-1][0] + 1
super().add_widget(marker, **kwargs)
def add_widget(self, marker):
marker._layer = self
self.markers.append(marker)
self.insert_marker(marker)
def remove_widget(self, marker):
marker._layer = None
if marker in self.markers:
self.markers.remove(marker)
super().remove_widget(marker)
def reposition(self):
if not self.markers:
return
mapview = self.parent
set_marker_position = self.set_marker_position
bbox = None
# reposition the markers depending the latitude
markers = sorted(self.markers, key=lambda x: -x.lat)
margin = max((max(marker.size) for marker in markers))
bbox = mapview.get_bbox(margin)
for marker in markers:
if bbox.collide(marker.lat, marker.lon):
set_marker_position(mapview, marker)
if not marker.parent:
self.insert_marker(marker)
else:
super().remove_widget(marker)
def set_marker_position(self, mapview, marker):
x, y = mapview.get_window_xy_from(marker.lat, marker.lon, mapview.zoom)
marker.x = int(x - marker.width * marker.anchor_x)
marker.y = int(y - marker.height * marker.anchor_y)
def unload(self):
self.clear_widgets()
del self.markers[:]
class MapViewScatter(Scatter):
# internal
def on_transform(self, *args):
super().on_transform(*args)
self.parent.on_transform(self.transform)
def collide_point(self, x, y):
return True
class MapView(Widget):
"""MapView is the widget that control the map displaying, navigation, and
layers management.
"""
lon = NumericProperty()
"""Longitude at the center of the widget
"""
lat = NumericProperty()
"""Latitude at the center of the widget
"""
zoom = NumericProperty(0)
"""Zoom of the widget. Must be between :meth:`MapSource.get_min_zoom` and
:meth:`MapSource.get_max_zoom`. Default to 0.
"""
map_source = ObjectProperty(MapSource())
"""Provider of the map, default to a empty :class:`MapSource`.
"""
double_tap_zoom = BooleanProperty(False)
"""If True, this will activate the double-tap to zoom.
"""
pause_on_action = BooleanProperty(True)
"""Pause any map loading / tiles loading when an action is done.
This allow better performance on mobile, but can be safely deactivated on
desktop.
"""
snap_to_zoom = BooleanProperty(True)
"""When the user initiate a zoom, it will snap to the closest zoom for
better graphics. The map can be blur if the map is scaled between 2 zoom.
Default to True, even if it doesn't fully working yet.
"""
animation_duration = NumericProperty(100)
"""Duration to animate Tiles alpha from 0 to 1 when it's ready to show.
Default to 100 as 100ms. Use 0 to deactivate.
"""
delta_x = NumericProperty(0)
delta_y = NumericProperty(0)
background_color = ListProperty([181 / 255.0, 208 / 255.0, 208 / 255.0, 1])
cache_dir = StringProperty(CACHE_DIR)
_zoom = NumericProperty(0)
_pause = BooleanProperty(False)
_scale = 1.0
_disabled_count = 0
__events__ = ["on_map_relocated"]
# Public API
@property
def viewport_pos(self):
vx, vy = self._scatter.to_local(self.x, self.y)
return vx - self.delta_x, vy - self.delta_y
@property
def scale(self):
if self._invalid_scale:
self._invalid_scale = False
self._scale = self._scatter.scale
return self._scale
def get_bbox(self, margin=0):
"""Returns the bounding box from the bottom/left (lat1, lon1) to
top/right (lat2, lon2).
"""
x1, y1 = self.to_local(0 - margin, 0 - margin)
x2, y2 = self.to_local((self.width + margin), (self.height + margin))
c1 = self.get_latlon_at(x1, y1)
c2 = self.get_latlon_at(x2, y2)
return Bbox((c1.lat, c1.lon, c2.lat, c2.lon))
bbox = AliasProperty(get_bbox, None, bind=["lat", "lon", "_zoom"])
def unload(self):
"""Unload the view and all the layers.
It also cancel all the remaining downloads.
"""
self.remove_all_tiles()
def get_window_xy_from(self, lat, lon, zoom):
"""Returns the x/y position in the widget absolute coordinates
from a lat/lon"""
scale = self.scale
vx, vy = self.viewport_pos
ms = self.map_source
x = ms.get_x(zoom, lon) - vx
y = ms.get_y(zoom, lat) - vy
x *= scale
y *= scale
x = x + self.pos[0]
y = y + self.pos[1]
return x, y
def center_on(self, *args):
"""Center the map on the coordinate :class:`Coordinate`, or a (lat, lon)
"""
map_source = self.map_source
zoom = self._zoom
if len(args) == 1 and isinstance(args[0], Coordinate):
coord = args[0]
lat = coord.lat
lon = coord.lon
elif len(args) == 2:
lat, lon = args
else:
raise Exception("Invalid argument for center_on")
lon = clamp(lon, MIN_LONGITUDE, MAX_LONGITUDE)
lat = clamp(lat, MIN_LATITUDE, MAX_LATITUDE)
scale = self._scatter.scale
x = map_source.get_x(zoom, lon) - self.center_x / scale
y = map_source.get_y(zoom, lat) - self.center_y / scale
self.delta_x = -x
self.delta_y = -y
self.lon = lon
self.lat = lat
self._scatter.pos = 0, 0
self.trigger_update(True)
def set_zoom_at(self, zoom, x, y, scale=None):
"""Sets the zoom level, leaving the (x, y) at the exact same point
in the view.
"""
zoom = clamp(
zoom, self.map_source.get_min_zoom(), self.map_source.get_max_zoom()
)
if int(zoom) == int(self._zoom):
if scale is None:
return
elif scale == self.scale:
return
scale = scale or 1.0
# first, rescale the scatter
scatter = self._scatter
scale = clamp(scale, scatter.scale_min, scatter.scale_max)
rescale = scale * 1.0 / scatter.scale
scatter.apply_transform(
Matrix().scale(rescale, rescale, rescale),
post_multiply=True,
anchor=scatter.to_local(x, y),
)
# adjust position if the zoom changed
c1 = self.map_source.get_col_count(self._zoom)
c2 = self.map_source.get_col_count(zoom)
if c1 != c2:
f = float(c2) / float(c1)
self.delta_x = scatter.x + self.delta_x * f
self.delta_y = scatter.y + self.delta_y * f
# back to 0 every time
scatter.apply_transform(
Matrix().translate(-scatter.x, -scatter.y, 0), post_multiply=True
)
# avoid triggering zoom changes.
self._zoom = zoom
self.zoom = self._zoom
def on_zoom(self, instance, zoom):
if zoom == self._zoom:
return
x = self.map_source.get_x(zoom, self.lon) - self.delta_x
y = self.map_source.get_y(zoom, self.lat) - self.delta_y
self.set_zoom_at(zoom, x, y)
self.center_on(self.lat, self.lon)
def get_latlon_at(self, x, y, zoom=None):
"""Return the current :class:`Coordinate` within the (x, y) widget
coordinate.
"""
if zoom is None:
zoom = self._zoom
vx, vy = self.viewport_pos
scale = self._scale
return Coordinate(
lat=self.map_source.get_lat(zoom, y / scale + vy),
lon=self.map_source.get_lon(zoom, x / scale + vx),
)
def add_marker(self, marker, layer=None):
"""Add a marker into the layer. If layer is None, it will be added in
the default marker layer. If there is no default marker layer, a new
one will be automatically created
"""
if layer is None:
if not self._default_marker_layer:
layer = MarkerMapLayer()
self.add_layer(layer)
else:
layer = self._default_marker_layer
layer.add_widget(marker)
layer.set_marker_position(self, marker)
def remove_marker(self, marker):
"""Remove a marker from its layer
"""
marker.detach()
def add_layer(self, layer, mode="window"):
"""Add a new layer to update at the same time the base tile layer.
mode can be either "scatter" or "window". If "scatter", it means the
layer will be within the scatter transformation. It's perfect if you
want to display path / shape, but not for text.
If "window", it will have no transformation. You need to position the
widget yourself: think as Z-sprite / billboard.
Defaults to "window".
"""
assert mode in ("scatter", "window")
if self._default_marker_layer is None and isinstance(layer, MarkerMapLayer):
self._default_marker_layer = layer
self._layers.append(layer)
c = self.canvas
if mode == "scatter":
self.canvas = self.canvas_layers
else:
self.canvas = self.canvas_layers_out
layer.canvas_parent = self.canvas
super().add_widget(layer)
self.canvas = c
def remove_layer(self, layer):
"""Remove the layer
"""
c = self.canvas
self._layers.remove(layer)
self.canvas = layer.canvas_parent
super().remove_widget(layer)
self.canvas = c
def sync_to(self, other):
"""Reflect the lat/lon/zoom of the other MapView to the current one.
"""
if self._zoom != other._zoom:
self.set_zoom_at(other._zoom, *self.center)
self.center_on(other.get_latlon_at(*self.center))
# Private API
def __init__(self, **kwargs):
from kivy.base import EventLoop
EventLoop.ensure_window()
self._invalid_scale = True
self._tiles = []
self._tiles_bg = []
self._tilemap = {}
self._layers = []
self._default_marker_layer = None
self._need_redraw_all = False
self._transform_lock = False
self.trigger_update(True)
self.canvas = Canvas()
self._scatter = MapViewScatter()
self.add_widget(self._scatter)
with self._scatter.canvas:
self.canvas_map = Canvas()
self.canvas_layers = Canvas()
with self.canvas:
self.canvas_layers_out = Canvas()
self._scale_target_anim = False
self._scale_target = 1.0
self._touch_count = 0
self.map_source.cache_dir = self.cache_dir
Clock.schedule_interval(self._animate_color, 1 / 60.0)
self.lat = kwargs.get("lat", self.lat)
self.lon = kwargs.get("lon", self.lon)
super().__init__(**kwargs)
def _animate_color(self, dt):
# fast path
d = self.animation_duration
if d == 0:
for tile in self._tiles:
if tile.state == "need-animation":
tile.g_color.a = 1.0
tile.state = "animated"
for tile in self._tiles_bg:
if tile.state == "need-animation":
tile.g_color.a = 1.0
tile.state = "animated"
else:
d = d / 1000.0
for tile in self._tiles:
if tile.state != "need-animation":
continue
tile.g_color.a += dt / d
if tile.g_color.a >= 1:
tile.state = "animated"
for tile in self._tiles_bg:
if tile.state != "need-animation":
continue
tile.g_color.a += dt / d
if tile.g_color.a >= 1:
tile.state = "animated"
def add_widget(self, widget):
if isinstance(widget, MapMarker):
self.add_marker(widget)
elif isinstance(widget, MapLayer):
self.add_layer(widget)
else:
super().add_widget(widget)
def remove_widget(self, widget):
if isinstance(widget, MapMarker):
self.remove_marker(widget)
elif isinstance(widget, MapLayer):
self.remove_layer(widget)
else:
super().remove_widget(widget)
def on_map_relocated(self, zoom, coord):
pass
def animated_diff_scale_at(self, d, x, y):
self._scale_target_time = 1.0
self._scale_target_pos = x, y
if self._scale_target_anim is False:
self._scale_target_anim = True
self._scale_target = d
else:
self._scale_target += d
Clock.unschedule(self._animate_scale)
Clock.schedule_interval(self._animate_scale, 1 / 60.0)
def _animate_scale(self, dt):
diff = self._scale_target / 3.0
if abs(diff) < 0.01:
diff = self._scale_target
self._scale_target = 0
else:
self._scale_target -= diff
self._scale_target_time -= dt
self.diff_scale_at(diff, *self._scale_target_pos)
ret = self._scale_target != 0
if not ret:
self._pause = False
return ret
def diff_scale_at(self, d, x, y):
scatter = self._scatter
scale = scatter.scale * (2 ** d)
self.scale_at(scale, x, y)
def scale_at(self, scale, x, y):
scatter = self._scatter
scale = clamp(scale, scatter.scale_min, scatter.scale_max)
rescale = scale * 1.0 / scatter.scale
scatter.apply_transform(
Matrix().scale(rescale, rescale, rescale),
post_multiply=True,
anchor=scatter.to_local(x, y),
)
def on_touch_down(self, touch):
if not self.collide_point(*touch.pos):
return
if self.pause_on_action:
self._pause = True
if "button" in touch.profile and touch.button in ("scrolldown", "scrollup"):
d = 1 if touch.button == "scrollup" else -1
self.animated_diff_scale_at(d, *touch.pos)
return True
elif touch.is_double_tap and self.double_tap_zoom:
self.animated_diff_scale_at(1, *touch.pos)
return True
touch.grab(self)
self._touch_count += 1
if self._touch_count == 1:
self._touch_zoom = (self.zoom, self._scale)
return super().on_touch_down(touch)
def on_touch_up(self, touch):
if touch.grab_current == self:
touch.ungrab(self)
self._touch_count -= 1
if self._touch_count == 0:
# animate to the closest zoom
zoom, scale = self._touch_zoom
cur_zoom = self.zoom
cur_scale = self._scale
if cur_zoom < zoom or cur_scale < scale:
self.animated_diff_scale_at(1.0 - cur_scale, *touch.pos)
elif cur_zoom > zoom or cur_scale > scale:
self.animated_diff_scale_at(2.0 - cur_scale, *touch.pos)
self._pause = False
return True
return super().on_touch_up(touch)
def on_transform(self, *args):
self._invalid_scale = True
if self._transform_lock:
return
self._transform_lock = True
# recalculate viewport
map_source = self.map_source
zoom = self._zoom
scatter = self._scatter
scale = scatter.scale
if scale >= 2.0:
zoom += 1
scale /= 2.0
elif scale < 1:
zoom -= 1
scale *= 2.0
zoom = clamp(zoom, map_source.min_zoom, map_source.max_zoom)
if zoom != self._zoom:
self.set_zoom_at(zoom, scatter.x, scatter.y, scale=scale)
self.trigger_update(True)
else:
if zoom == map_source.min_zoom and scatter.scale < 1.0:
scatter.scale = 1.0
self.trigger_update(True)
else:
self.trigger_update(False)
if map_source.bounds:
self._apply_bounds()
self._transform_lock = False
self._scale = self._scatter.scale
def _apply_bounds(self):
# if the map_source have any constraints, apply them here.
map_source = self.map_source
zoom = self._zoom
min_lon, min_lat, max_lon, max_lat = map_source.bounds
xmin = map_source.get_x(zoom, min_lon)
xmax = map_source.get_x(zoom, max_lon)
ymin = map_source.get_y(zoom, min_lat)
ymax = map_source.get_y(zoom, max_lat)
dx = self.delta_x
dy = self.delta_y
oxmin, oymin = self._scatter.to_local(self.x, self.y)
oxmax, oymax = self._scatter.to_local(self.right, self.top)
s = self._scale
cxmin = oxmin - dx
if cxmin < xmin:
self._scatter.x += (cxmin - xmin) * s
cymin = oymin - dy
if cymin < ymin:
self._scatter.y += (cymin - ymin) * s
cxmax = oxmax - dx
if cxmax > xmax:
self._scatter.x -= (xmax - cxmax) * s
cymax = oymax - dy
if cymax > ymax:
self._scatter.y -= (ymax - cymax) * s
def on__pause(self, instance, value):
if not value:
self.trigger_update(True)
def trigger_update(self, full):
self._need_redraw_full = full or self._need_redraw_full
Clock.unschedule(self.do_update)
Clock.schedule_once(self.do_update, -1)
def do_update(self, dt):
zoom = self._zoom
scale = self._scale
self.lon = self.map_source.get_lon(
zoom, (self.center_x - self._scatter.x) / scale - self.delta_x
)
self.lat = self.map_source.get_lat(
zoom, (self.center_y - self._scatter.y) / scale - self.delta_y
)
self.dispatch("on_map_relocated", zoom, Coordinate(self.lon, self.lat))
for layer in self._layers:
layer.reposition()
if self._need_redraw_full:
self._need_redraw_full = False
self.move_tiles_to_background()
self.load_visible_tiles()
else:
self.load_visible_tiles()
def bbox_for_zoom(self, vx, vy, w, h, zoom):
# return a tile-bbox for the zoom
map_source = self.map_source
size = map_source.dp_tile_size
scale = self._scale
max_x_end = map_source.get_col_count(zoom)
max_y_end = map_source.get_row_count(zoom)
x_count = int(ceil(w / scale / float(size))) + 1
y_count = int(ceil(h / scale / float(size))) + 1
tile_x_first = int(clamp(vx / float(size), 0, max_x_end))
tile_y_first = int(clamp(vy / float(size), 0, max_y_end))
tile_x_last = tile_x_first + x_count
tile_y_last = tile_y_first + y_count
tile_x_last = int(clamp(tile_x_last, tile_x_first, max_x_end))
tile_y_last = int(clamp(tile_y_last, tile_y_first, max_y_end))
x_count = tile_x_last - tile_x_first
y_count = tile_y_last - tile_y_first
return (tile_x_first, tile_y_first, tile_x_last, tile_y_last, x_count, y_count)
def load_visible_tiles(self):
map_source = self.map_source
vx, vy = self.viewport_pos
zoom = self._zoom
dirs = [0, 1, 0, -1, 0]
bbox_for_zoom = self.bbox_for_zoom
size = map_source.dp_tile_size
(
tile_x_first,
tile_y_first,
tile_x_last,
tile_y_last,
x_count,
y_count,
) = bbox_for_zoom(vx, vy, self.width, self.height, zoom)
# Adjust tiles behind us
for tile in self._tiles_bg[:]:
tile_x = tile.tile_x
tile_y = tile.tile_y
f = 2 ** (zoom - tile.zoom)
w = self.width / f
h = self.height / f
(
btile_x_first,
btile_y_first,
btile_x_last,
btile_y_last,
_,
_,
) = bbox_for_zoom(vx / f, vy / f, w, h, tile.zoom)
if (
tile_x < btile_x_first
or tile_x >= btile_x_last
or tile_y < btile_y_first
or tile_y >= btile_y_last
):
tile.state = "done"
self._tiles_bg.remove(tile)
self.canvas_map.before.remove(tile.g_color)
self.canvas_map.before.remove(tile)
continue
tsize = size * f
tile.size = tsize, tsize
tile.pos = (tile_x * tsize + self.delta_x, tile_y * tsize + self.delta_y)
# Get rid of old tiles first
for tile in self._tiles[:]:
tile_x = tile.tile_x
tile_y = tile.tile_y
if (
tile_x < tile_x_first
or tile_x >= tile_x_last
or tile_y < tile_y_first
or tile_y >= tile_y_last
):
tile.state = "done"
self.tile_map_set(tile_x, tile_y, False)
self._tiles.remove(tile)
self.canvas_map.remove(tile)
self.canvas_map.remove(tile.g_color)
else:
tile.size = (size, size)
tile.pos = (tile_x * size + self.delta_x, tile_y * size + self.delta_y)
# Load new tiles if needed
x = tile_x_first + x_count // 2 - 1
y = tile_y_first + y_count // 2 - 1
arm_max = max(x_count, y_count) + 2
arm_size = 1
turn = 0
while arm_size < arm_max:
for i in range(arm_size):
if (
not self.tile_in_tile_map(x, y)
and y >= tile_y_first
and y < tile_y_last
and x >= tile_x_first
and x < tile_x_last
):
self.load_tile(x, y, size, zoom)
x += dirs[turn % 4 + 1]
y += dirs[turn % 4]
if turn % 2 == 1:
arm_size += 1
turn += 1
def load_tile(self, x, y, size, zoom):
if self.tile_in_tile_map(x, y) or zoom != self._zoom:
return
self.load_tile_for_source(self.map_source, 1.0, size, x, y, zoom)
# XXX do overlay support
self.tile_map_set(x, y, True)
def load_tile_for_source(self, map_source, opacity, size, x, y, zoom):
tile = Tile(size=(size, size), cache_dir=self.cache_dir)
tile.g_color = Color(1, 1, 1, 0)
tile.tile_x = x
tile.tile_y = y
tile.zoom = zoom
tile.pos = (x * size + self.delta_x, y * size + self.delta_y)
tile.map_source = map_source
tile.state = "loading"
if not self._pause:
map_source.fill_tile(tile)
self.canvas_map.add(tile.g_color)
self.canvas_map.add(tile)
self._tiles.append(tile)
def move_tiles_to_background(self):
# remove all the tiles of the main map to the background map
# retain only the one who are on the current zoom level
# for all the tile in the background, stop the download if not yet started.
zoom = self._zoom
tiles = self._tiles
btiles = self._tiles_bg
canvas_map = self.canvas_map
tile_size = self.map_source.tile_size
# move all tiles to background
while tiles:
tile = tiles.pop()
if tile.state == "loading":
tile.state = "done"
continue
btiles.append(tile)
# clear the canvas
canvas_map.clear()
canvas_map.before.clear()
self._tilemap = {}
# unsure if it's really needed, i personnally didn't get issues right now
# btiles.sort(key=lambda z: -z.zoom)
# add all the btiles into the back canvas.
# except for the tiles that are owned by the current zoom level
for tile in btiles[:]:
if tile.zoom == zoom:
btiles.remove(tile)
tiles.append(tile)
tile.size = tile_size, tile_size
canvas_map.add(tile.g_color)
canvas_map.add(tile)
self.tile_map_set(tile.tile_x, tile.tile_y, True)
continue
canvas_map.before.add(tile.g_color)
canvas_map.before.add(tile)
def remove_all_tiles(self):
# clear the map of all tiles.
self.canvas_map.clear()
self.canvas_map.before.clear()
for tile in self._tiles:
tile.state = "done"
del self._tiles[:]
del self._tiles_bg[:]
self._tilemap = {}
def tile_map_set(self, tile_x, tile_y, value):
key = tile_y * self.map_source.get_col_count(self._zoom) + tile_x
if value:
self._tilemap[key] = value
else:
self._tilemap.pop(key, None)
def tile_in_tile_map(self, tile_x, tile_y):
key = tile_y * self.map_source.get_col_count(self._zoom) + tile_x
return key in self._tilemap
def on_size(self, instance, size):
for layer in self._layers:
layer.size = size
self.center_on(self.lat, self.lon)
self.trigger_update(True)
def on_pos(self, instance, pos):
self.center_on(self.lat, self.lon)
self.trigger_update(True)
def on_map_source(self, instance, source):
if isinstance(source, string_types):
self.map_source = MapSource.from_provider(source)
elif isinstance(source, (tuple, list)):
cache_key, min_zoom, max_zoom, url, attribution, options = source
self.map_source = MapSource(
url=url,
cache_key=cache_key,
min_zoom=min_zoom,
max_zoom=max_zoom,
attribution=attribution,
cache_dir=self.cache_dir,
**options
)
elif isinstance(source, MapSource):
self.map_source = source
else:
raise Exception("Invalid map source provider")
self.zoom = clamp(self.zoom, self.map_source.min_zoom, self.map_source.max_zoom)
self.remove_all_tiles()
self.trigger_update(True)

16
kivyblocks/register.py
View File

@ -16,11 +16,25 @@ from .qrdata import QRCodeWidget
# from .kivycamera import KivyCamera
from .filebrowser import FileLoaderBrowser
from .osc_server import OSCServer
from graph import Graph, MeshLinePlot, MeshStemPlot, LinePlot, \
from .graph import Graph, MeshLinePlot, MeshStemPlot, LinePlot, \
SmoothLinePlot, ContourPlot, BarPlot, HBar, VBar, ScatterPlot, \
PointPlot
from .mapview import MapView
r = Factory.register
r('Popup', Popup)
r('Graph', Graph)
r('MeshLinePlot', MeshLinePlot)
r('MeshStemPlot', MeshStemPlot)
r('LinePlot', LinePlot)
r('SmoothLinePlot', SmoothLinePlot)
r('ContourPlot', ContourPlot)
r('BarPlot', BarPlot)
r('HBar', HBar)
r('VBar', VBar)
r('ScatterPlot', ScatterPlot)
r('PointPlot', PointPlot)
r('MapView', MapView)
r('OSCServer',OSCServer)
r('DataGrid',DataGrid)
r('FileLoaderBrowser',FileLoaderBrowser)

34
kivyblocks/responsivelayout.py
View File

@ -14,23 +14,14 @@ class VResponsiveLayout(ScrollView):
self.box_cols = cols
super(VResponsiveLayout, self).__init__(**kw)
self.options = kw
print('VResponsiveLayout():cols=',self.org_cols,'box_width=',self.box_width)
self._inner = GridLayout(cols=self.org_cols, padding=2,
spacing=2,size_hint=(1,None))
super(VResponsiveLayout,self).add_widget(self._inner)
self._inner.bind(
minimum_height=self._inner.setter('height'))
self.sizeChangedTask = None
self.bind(pos=self.sizeChanged,size=self.sizeChanged)
self.bind(pos=self.setCols,size=self.setCols)
def sizeChanged(self,o,v=None):
if self.sizeChangedTask:
self.sizeChangedTask.cancel()
self.sizeChangedTask = Clock.schedule_once(self.sizeChangedWork,0.1)
def sizeChangedWork(self,t=None):
self.setCols()
def on_orientation(self,o):
self.setCols()
@ -49,19 +40,14 @@ class VResponsiveLayout(ScrollView):
a = self._inner.remove_widget(widget,**kw)
return a
def setCols(self,t=None):
self.box_width = self.org_box_width
if self.width < self.box_width:
self.cols = self.org_cols
else:
self.cols = int(self.width / self.box_width)
print('VResponsiveLayout()::setCols():self.cols=',self.cols, 'self.box_width=',self.box_width,'self.org_cols=',self.org_cols)
if isHandHold():
w,h = self.size
if w < h:
self.box_width = w / self.org_cols - 2
self.cols = self.org_cols
self._inner.cols = self.cols
def setCols(self,*args):
cols = round(self.width / self.org_box_width)
if cols < 1:
return
if isHandHold() and self.width < self.height:
cols = self.org_cols
box_width = self.width / cols - 2
self._inner.cols = cols
for w in self._inner.children:
w.width = self.box_width
w.width = box_width

13
kivyblocks/toggleitems.py
View File

@ -36,6 +36,17 @@ class PressableBox(TouchRippleButtonBehavior, Box):
def getValue(self):
return self.user_data
"""
ToggleItems format:
{
color_level:
radius:
unit_size:
items_desc:
border_width:
orientation:
}
"""
class ToggleItems(BGColorBehavior, BoxLayout):
def __init__(self,
color_level=1,
@ -112,6 +123,6 @@ class ToggleItems(BGColorBehavior, BoxLayout):
for i in self.item_widgets:
if i.getValue() == v:
i.selected()
self.select_iten(i)
self.select_item(i)
return

223
kivyblocks/toolbar.py
View File

@ -16,84 +16,16 @@ from .ready import WidgetReady
from .color_definitions import getColors
from .bgcolorbehavior import BGColorBehavior
from .baseWidget import Text
"""
toobar={
"mode":"icon", "icontext","text"
img_size=2,
text_size=1,
"tools":[
]
}
tool options
{
name:'',
label:''
img=''
}
"""
class Tool(ButtonBehavior, BGColorBehavior, BoxLayout):
def __init__(self,ancestor=None,**opts):
if ancestor is None:
ancestor = App.get_running_app().root
self.widget_id = opts['name']
ButtonBehavior.__init__(self)
BoxLayout.__init__(self,
size_hint_y=None)
BGColorBehavior.__init__(self,color_level=ancestor.color_level,
radius=ancestor.radius)
self.bl = BoxLayout(orientation='vertical',
size_hint_y=None)
self.add_widget(self.bl)
self.opts = DictObject(**opts)
if not self.opts.img_size:
self.opts.img_size = 2
if not self.opts.text_size:
self.opts.text_size = 1
app = App.get_running_app()
size = 0
if self.opts.img_src:
size = CSize(self.opts.img_size or 2)
img = AsyncImage(source=self.opts.img_src,
size_hint=(None,None),
size=(size,size))
self.bl.add_widget(img)
tsize = CSize(self.opts.text_size)
label = self.opts.label or self.opts.name
self.lbl = Text(i18n=True,
text=label,
font_size=int(tsize),
text_size=(CSize(len(label)), tsize),
height=tsize,
width=CSize(len(label)),
size_hint=(None,None),
)
self.bl.add_widget(self.lbl)
self.height = (size + tsize)*1.1
def on_size(self,obj,size):
Logger.info('toolbar: Tool() on_size fired')
#self.lbl.color, self.bgcolor = getColors(self.ancestor.color_level,
# selected=False)
#self.lbl.bgcolor = self.bgcolor
def on_press(self):
print('Tool(). pressed ...............')
def setActive(self,flag):
if flag:
self.selected()
else:
self.unselected()
from .toggleitems import PressableBox, ToggleItems
"""
toolbar options
{
color_level:
radius:
"mode":"icon", "icontext","text"
img_size=1.5,
text_size=0.7,
text_size=0.5,
tools:[
{
"name":"myid",
@ -104,44 +36,73 @@ toolbar options
]
}
"""
class Toolbar(BGColorBehavior, GridLayout):
def __init__(self, ancestor=None,**opts):
self.opts = DictObject(**opts)
class Toolbar(BoxLayout):
def __init__(self, color_level=-1,
radius=[],
img_size=1.5,
text_size=0.5,
tools=[], **opts):
self.color_level = color_level
self.radius = radius
self.img_size = img_size
self.text_size = text_size
self.tools = tools
self.tool_widgets={}
GridLayout.__init__(self,cols = len(self.opts.tools))
color_level = 0
if isinstance(ancestor, BGColorBehavior):
color_level = ancestor.color_level + 1
BGColorBehavior.__init__(self,color_level=color_level)
BoxLayout.__init__(self, **opts)
self.register_event_type('on_press')
self.size_hint = (1,None)
first = True
for opt in self.opts.tools:
opt.img_size = self.opts.img_size
opt.text_size = self.opts.text_size
if opt.img_src:
purl = None
if ancestor and hasattr(ancestor, 'parenturl'):
purl = ancestor.parenturl
opt.img_src = absurl(opt.img_src,purl)
tool = Tool(ancestor=ancestor, **opt)
if first:
first = False
h = ancestor
if not ancestor:
h = App.get_runnung_app().root
self.tool_widgets[opt.name] = tool
box = BoxLayout()
box.add_widget(tool)
self.add_widget(box)
tool.bind(on_press=self.tool_press)
self.height = tool.height * 1.1
subs_desc = []
for opt in self.tools:
subwidgets = []
img_src = opt.get('img_src',None)
if img_src:
subwidgets.append({
"widgettype":"AsyncImage",
"options":{
"size_hint_y":None,
"height":CSize(self.img_size),
"source":img_src
}
})
text = opt.get('label', None)
if text:
subwidgets.append({
"widgettype":"Text",
"options":{
"size_hint_y":None,
"i18n":True,
"height":CSize(self.text_size),
"font_size":CSize(self.text_size),
"text":text
}
})
desc = {
"widgettype":"VBox",
"options":{
},
"subwidgets":subwidgets,
"data":opt.get('name')
}
subs_desc.append(desc)
self.toggle_items = ToggleItems(
color_level=self.color_level,
radius=self.radius,
unit_size=self.img_size + self.text_size,
items_desc=subs_desc)
for ti in self.toggle_items.children:
ti.widget_id = ti.user_data
self.height = CSize(self.img_size + self.text_size) + 10
self.size_hint_y = None
self.toggle_items.bind(on_press=self.tool_press)
self.add_widget(self.toggle_items)
def on_press(self, o):
print('on_press(),', o)
def tool_press(self,o,v=None):
for n,w in self.tool_widgets.items():
active = False
if w == o:
active = True
w.setActive(active)
self.dispatch('on_press',self.toggle_items.getValue())
"""
Toolpage options
@ -162,53 +123,63 @@ Toolpage options
"""
class ToolPage(BGColorBehavior, BoxLayout):
def __init__(self,color_level=-1,radius=[],**opts):
def __init__(self,color_level=-1,radius=[],tool_at='top', **opts):
self.opts = DictObject(**opts)
if self.opts.tool_at in [ 'top','bottom']:
if tool_at in [ 'top','bottom']:
orient = 'vertical'
else:
orient = 'horizontal'
self.color_level=self.opts.color_level or 0
self.radius = self.opts.radius
self.sub_radius = self.opts.radius
self.tool_at = tool_at
BoxLayout.__init__(self,orientation=orient)
BGColorBehavior.__init__(self,
color_level=color_level,
radius=radius)
radius=[])
self.content = None
self.toolbar = None
self.init()
self.show_firstpage()
# self.show_firstpage()
def on_size(self,obj,size):
if self.content is None:
return
x,y = size
self.toolbar.width = x
self.content.width = x
self.content.height = y - self.toolbar.height
if self.tool_at in ['top','bottom']:
self.toolbar.width = self.width
self.content.width = self.width
self.content.height = self.height - self.toolbar.height
else:
self.toolbar.height = self.height
self.content.height = self.height
self.content.width = self.width - self.toolbar.width
def showPage(self,obj):
self._show_page(obj.opts)
def show_firstpage(self,t=None):
return
d = self.children[0]
d.dispatch('on_press')
d.dispatch('on_press', d)
def init(self):
self.initFlag = True
self.mywidgets = {}
self.content = BoxLayout()
self.content.widget_id = 'content'
for t in self.opts.tools:
parenturl = None
if hasattr(self,'parenturl'):
parenturl = self.parenturl
t.img_src = absurl(t.img_src,parenturl)
opts = self.opts
self.toolbar = Toolbar(ancestor=self, **self.opts)
if self.opts.tool_at in ['top','left']:
opts = self.opts.copy()
if self.tool_at in ['top','bottom']:
opts['size_hint_x'] = 1
opts['size_hint_y'] = None
opts['height'] = CSize(self.opts.img_size + \
self.opts.text_size) + 10
else:
opts['size_hint_y'] = 1
opts['size_hint_x'] = None
opts['width'] = CSize(self.opts.img_size + \
self.opts.text_size) + 10
self.toolbar = Toolbar(color_level=self.color_level,
radius=self.sub_radius,
**opts)
if self.tool_at in ['top','left']:
self.add_widget(self.toolbar)
self.add_widget(self.content)
else: