536 lines
20 KiB
Python
536 lines
20 KiB
Python
"""Tkinter-based interactive demo for python-map-manager.
|
|
|
|
Provides a small GUI to exercise MapEngine and MapVisualizer with preset
|
|
examples and controls for point/area fetching, caching and saving images.
|
|
"""
|
|
import sys
|
|
import pathlib
|
|
import os
|
|
import threading
|
|
import time
|
|
from typing import Optional
|
|
|
|
script_dir = pathlib.Path(__file__).parent.resolve()
|
|
if str(script_dir) not in sys.path:
|
|
sys.path.insert(0, str(script_dir))
|
|
|
|
try:
|
|
import tkinter as tk
|
|
from tkinter import ttk, filedialog, messagebox
|
|
except Exception:
|
|
tk = None
|
|
|
|
from PIL import Image, ImageTk
|
|
|
|
from map_manager.engine import MapEngine
|
|
from map_manager.visualizer import MapVisualizer
|
|
|
|
|
|
class DebugGUI:
|
|
def __init__(self, root):
|
|
self.root = root
|
|
root.title('python-map-manager Demo')
|
|
|
|
self.service = 'osm'
|
|
self.cache_dir = 'debug_map_cache'
|
|
self.online = True
|
|
self.engine = MapEngine(service_name=self.service, cache_dir=self.cache_dir, enable_online=self.online)
|
|
self.visual = MapVisualizer(self.engine)
|
|
self.last_image = None
|
|
self._photo = None
|
|
|
|
self._build_ui()
|
|
|
|
def _build_ui(self):
|
|
frm = ttk.Frame(self.root, padding=8)
|
|
frm.grid(column=0, row=0, sticky='nsew')
|
|
self.root.columnconfigure(0, weight=1)
|
|
self.root.rowconfigure(0, weight=1)
|
|
|
|
# Left: controls
|
|
controls = ttk.Frame(frm)
|
|
controls.grid(column=0, row=0, sticky='nw')
|
|
|
|
# Point inputs
|
|
ttk.Label(controls, text='Point (lat lon zoom radius)').grid(column=0, row=0, sticky='w')
|
|
self.point_lat = tk.StringVar(value='41.9028')
|
|
self.point_lon = tk.StringVar(value='12.4964')
|
|
self.point_zoom = tk.StringVar(value='12')
|
|
self.point_radius = tk.StringVar(value='1')
|
|
e_lat = ttk.Entry(controls, width=10, textvariable=self.point_lat)
|
|
e_lon = ttk.Entry(controls, width=10, textvariable=self.point_lon)
|
|
e_zoom = ttk.Entry(controls, width=6, textvariable=self.point_zoom)
|
|
e_rad = ttk.Entry(controls, width=6, textvariable=self.point_radius)
|
|
e_lat.grid(column=0, row=1)
|
|
e_lon.grid(column=1, row=1)
|
|
e_zoom.grid(column=2, row=1)
|
|
e_rad.grid(column=3, row=1)
|
|
self.btn_point = ttk.Button(controls, text='Fetch Point', command=self.fetch_point)
|
|
self.btn_point.grid(column=0, row=2, columnspan=2, pady=4)
|
|
|
|
# Area inputs
|
|
ttk.Label(controls, text='Area (W S E N)').grid(column=0, row=3, sticky='w', pady=(8,0))
|
|
self.area_w = tk.StringVar(value='12.49')
|
|
self.area_s = tk.StringVar(value='41.89')
|
|
self.area_e = tk.StringVar(value='12.50')
|
|
self.area_n = tk.StringVar(value='41.90')
|
|
a_w = ttk.Entry(controls, width=10, textvariable=self.area_w)
|
|
a_s = ttk.Entry(controls, width=10, textvariable=self.area_s)
|
|
a_e = ttk.Entry(controls, width=10, textvariable=self.area_e)
|
|
a_n = ttk.Entry(controls, width=10, textvariable=self.area_n)
|
|
a_w.grid(column=0, row=4)
|
|
a_s.grid(column=1, row=4)
|
|
a_e.grid(column=2, row=4)
|
|
a_n.grid(column=3, row=4)
|
|
|
|
ttk.Label(controls, text='Zoom / MaxSize').grid(column=0, row=5, sticky='w')
|
|
self.area_zoom = tk.StringVar(value='')
|
|
self.area_maxsize = tk.StringVar(value='800')
|
|
az = ttk.Entry(controls, width=6, textvariable=self.area_zoom)
|
|
am = ttk.Entry(controls, width=8, textvariable=self.area_maxsize)
|
|
az.grid(column=0, row=6)
|
|
am.grid(column=1, row=6)
|
|
self.btn_area = ttk.Button(controls, text='Fetch Area', command=self.fetch_area)
|
|
self.btn_area.grid(column=0, row=7, columnspan=2, pady=4)
|
|
|
|
# Preset examples
|
|
ttk.Label(controls, text='Examples').grid(column=0, row=8, sticky='w', pady=(8,0))
|
|
self.examples = ttk.Combobox(controls, values=['Rome point', 'Rome area', 'Antimeridian area (toy)'])
|
|
self.examples.current(0)
|
|
self.examples.grid(column=0, row=9, columnspan=2, sticky='we')
|
|
ttk.Button(controls, text='Run Example', command=self.run_example).grid(column=2, row=9)
|
|
|
|
# Actions
|
|
ttk.Button(controls, text='Toggle Online', command=self.toggle_online).grid(column=0, row=10, pady=(12,0))
|
|
ttk.Button(controls, text='Clear Cache', command=self.clear_cache).grid(column=1, row=10, pady=(12,0))
|
|
ttk.Button(controls, text='Save Image', command=self.save_image).grid(column=2, row=10, pady=(12,0))
|
|
ttk.Button(controls, text='Show Image', command=self.show_image).grid(column=3, row=10, pady=(12,0))
|
|
|
|
# Status/log
|
|
self.status = tk.StringVar(value='Ready')
|
|
ttk.Label(controls, textvariable=self.status).grid(column=0, row=11, columnspan=4, sticky='we', pady=(8,0))
|
|
|
|
# Progress bar + ETA
|
|
self.progress_var = tk.IntVar(value=0)
|
|
self.progress = ttk.Progressbar(controls, orient='horizontal', length=200, mode='determinate', variable=self.progress_var)
|
|
self.progress.grid(column=0, row=12, columnspan=3, sticky='we', pady=(6,0))
|
|
self.eta_var = tk.StringVar(value='')
|
|
ttk.Label(controls, textvariable=self.eta_var).grid(column=3, row=12, sticky='w', pady=(6,0))
|
|
|
|
# Right: image preview
|
|
preview = ttk.Frame(frm)
|
|
preview.grid(column=1, row=0, sticky='nsew', padx=(12,0))
|
|
frm.columnconfigure(1, weight=1)
|
|
frm.rowconfigure(0, weight=1)
|
|
|
|
self.canvas = tk.Label(preview, text='No image', background='black', width=60, height=30)
|
|
self.canvas.pack(fill='both', expand=True)
|
|
|
|
def log(self, msg: str):
|
|
print(msg)
|
|
self.status.set(msg)
|
|
|
|
def fetch_point(self):
|
|
try:
|
|
lat = float(self.point_lat.get())
|
|
lon = float(self.point_lon.get())
|
|
zoom = int(self.point_zoom.get())
|
|
radius = int(self.point_radius.get())
|
|
except Exception:
|
|
messagebox.showerror('Input error', 'Invalid numeric input for point')
|
|
return
|
|
self.log(f'Fetching point {lat},{lon} @ z={zoom} r={radius}...')
|
|
# run in background thread and show ETA based on per-tile durations
|
|
self._set_busy(True)
|
|
def worker():
|
|
durations = []
|
|
cached = 0
|
|
downloaded = 0
|
|
def progress_cb(done, total, last_dur, from_cache, tile_coords):
|
|
nonlocal cached, downloaded
|
|
durations.append(last_dur)
|
|
if from_cache:
|
|
cached += 1
|
|
else:
|
|
downloaded += 1
|
|
avg = sum(durations) / len(durations) if durations else 0.0
|
|
remaining = max(0, total - done)
|
|
eta = remaining * avg
|
|
# schedule UI update on main thread
|
|
try:
|
|
self.root.after(0, lambda: self._update_progress(done, total, eta, cached, downloaded))
|
|
except Exception:
|
|
pass
|
|
try:
|
|
img = self.engine.get_image_for_point(lat, lon, zoom, tiles_radius=radius, progress_callback=progress_cb)
|
|
if img is None:
|
|
self.root.after(0, lambda: self.log('No image returned (check dependencies and online state)'))
|
|
else:
|
|
self.last_image = img
|
|
self.root.after(0, lambda: self._update_preview(img))
|
|
self.root.after(0, lambda: self.log('Point image loaded'))
|
|
finally:
|
|
self.root.after(0, lambda: self._set_busy(False))
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
def fetch_area(self):
|
|
try:
|
|
w = float(self.area_w.get())
|
|
s = float(self.area_s.get())
|
|
e = float(self.area_e.get())
|
|
n = float(self.area_n.get())
|
|
except Exception:
|
|
messagebox.showerror('Input error', 'Invalid numeric input for area')
|
|
return
|
|
zoom = None
|
|
if self.area_zoom.get().strip():
|
|
try:
|
|
zoom = int(self.area_zoom.get())
|
|
except Exception:
|
|
messagebox.showerror('Input error', 'Invalid zoom')
|
|
return
|
|
maxsize = None
|
|
if self.area_maxsize.get().strip():
|
|
try:
|
|
maxsize = int(self.area_maxsize.get())
|
|
except Exception:
|
|
messagebox.showerror('Input error', 'Invalid maxsize')
|
|
return
|
|
self.log(f'Fetching area bbox=({w},{s},{e},{n}) zoom={zoom} maxsize={maxsize}...')
|
|
self._set_busy(True)
|
|
def worker():
|
|
durations = []
|
|
cached = 0
|
|
downloaded = 0
|
|
def progress_cb(done, total, last_dur, from_cache, tile_coords):
|
|
nonlocal cached, downloaded
|
|
durations.append(last_dur)
|
|
if from_cache:
|
|
cached += 1
|
|
else:
|
|
downloaded += 1
|
|
avg = sum(durations) / len(durations) if durations else 0.0
|
|
remaining = max(0, total - done)
|
|
eta = remaining * avg
|
|
try:
|
|
self.root.after(0, lambda: self._update_progress(done, total, eta, cached, downloaded))
|
|
except Exception:
|
|
pass
|
|
try:
|
|
img = self.engine.get_image_for_area((w, s, e, n), zoom=zoom, max_size=maxsize, progress_callback=progress_cb)
|
|
if img is None:
|
|
self.root.after(0, lambda: self.log('No image returned (check dependencies and online state)'))
|
|
else:
|
|
self.last_image = img
|
|
self.root.after(0, lambda: self._update_preview(img))
|
|
self.root.after(0, lambda: self.log('Area image loaded'))
|
|
finally:
|
|
self.root.after(0, lambda: self._set_busy(False))
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
def run_example(self):
|
|
sel = self.examples.get()
|
|
if sel == 'Rome point':
|
|
self.point_lat.set('41.9028')
|
|
self.point_lon.set('12.4964')
|
|
self.point_zoom.set('12')
|
|
self.point_radius.set('1')
|
|
self.fetch_point()
|
|
elif sel == 'Rome area':
|
|
self.area_w.set('12.49')
|
|
self.area_s.set('41.89')
|
|
self.area_e.set('12.50')
|
|
self.area_n.set('41.90')
|
|
self.area_maxsize.set('800')
|
|
self.fetch_area()
|
|
elif sel == 'Antimeridian area (toy)':
|
|
# small bbox crossing antimeridian near 179E/-179W
|
|
self.area_w.set('179.5')
|
|
self.area_s.set('-1.0')
|
|
self.area_e.set('-179.5')
|
|
self.area_n.set('1.0')
|
|
self.area_maxsize.set('600')
|
|
self.fetch_area()
|
|
|
|
def toggle_online(self):
|
|
self.online = not self.online
|
|
self.engine = MapEngine(service_name=self.service, cache_dir=self.cache_dir, enable_online=self.online)
|
|
self.visual = MapVisualizer(self.engine)
|
|
self.log(f'Online fetching set to {self.online}')
|
|
|
|
def clear_cache(self):
|
|
try:
|
|
self.engine.tile_manager.clear_entire_service_cache()
|
|
self.log('Cache cleared')
|
|
except Exception as e:
|
|
messagebox.showerror('Error', f'Failed to clear cache: {e}')
|
|
|
|
def save_image(self):
|
|
if not self.last_image:
|
|
messagebox.showinfo('No image', 'No image to save')
|
|
return
|
|
path = filedialog.asksaveasfilename(defaultextension='.png', filetypes=[('PNG','*.png')])
|
|
if not path:
|
|
return
|
|
try:
|
|
self.last_image.save(path)
|
|
self.log(f'Saved image to {path}')
|
|
except Exception as e:
|
|
messagebox.showerror('Save error', f'Failed to save: {e}')
|
|
|
|
def show_image(self):
|
|
if not self.last_image:
|
|
messagebox.showinfo('No image', 'No image to show')
|
|
return
|
|
try:
|
|
self.visual.show_pil_image(self.last_image)
|
|
except Exception as e:
|
|
messagebox.showerror('Show error', f'Failed to show image: {e}')
|
|
|
|
def _update_preview(self, pil_image: Image.Image):
|
|
# Resize image to fit preview area if needed
|
|
try:
|
|
w = max(200, min(1024, pil_image.width))
|
|
h = int(pil_image.height * (w / pil_image.width)) if pil_image.width else pil_image.height
|
|
thumb = pil_image.copy()
|
|
thumb.thumbnail((w, 800))
|
|
self._photo = ImageTk.PhotoImage(thumb)
|
|
self.canvas.configure(image=self._photo, text='')
|
|
except Exception as e:
|
|
self.canvas.configure(text=f'Preview error: {e}')
|
|
|
|
def _update_progress(self, done: int, total: int, eta_seconds: float, cached_count: int = 0, downloaded_count: int = 0):
|
|
try:
|
|
self.progress['maximum'] = total
|
|
self.progress_var.set(done)
|
|
if eta_seconds is None:
|
|
eta_text = ''
|
|
else:
|
|
eta_text = f'ETA: {int(eta_seconds)}s'
|
|
counts_text = f' D:{downloaded_count} C:{cached_count}'
|
|
self.eta_var.set(eta_text + counts_text)
|
|
except Exception:
|
|
pass
|
|
|
|
def _set_busy(self, busy: bool):
|
|
try:
|
|
state = 'disabled' if busy else 'normal'
|
|
self.btn_point.configure(state=state)
|
|
self.btn_area.configure(state=state)
|
|
if not busy:
|
|
# clear progress when done
|
|
self.progress_var.set(0)
|
|
self.progress['maximum'] = 1
|
|
self.eta_var.set('')
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main():
|
|
if tk is None:
|
|
print('Tkinter is not available in this Python environment.')
|
|
return
|
|
root = tk.Tk()
|
|
app = DebugGUI(root)
|
|
root.mainloop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
"""Simple debug tool to exercise MapEngine and MapVisualizer locally."""
|
|
"""Interactive debug tool for python-map-manager.
|
|
|
|
Provides a small REPL to exercise MapEngine and MapVisualizer without
|
|
editing code. Commands:
|
|
- point LAT LON ZOOM [RADIUS] -> fetch and show tiles around a point
|
|
- area W S E N [--zoom Z | --maxsize PIX] -> fetch and show area
|
|
- save PATH -> save last image to PATH
|
|
- show -> open last image with visualizer
|
|
- toggle_online -> toggle online fetching on/off
|
|
- clear_cache -> clear the map tiles cache for current service
|
|
- exit -> quit
|
|
|
|
This tool is intended for developer iteration and debugging.
|
|
"""
|
|
|
|
import sys
|
|
import pathlib
|
|
import os
|
|
import shlex
|
|
from typing import Optional
|
|
|
|
script_dir = pathlib.Path(__file__).parent.resolve()
|
|
if str(script_dir) not in sys.path:
|
|
sys.path.insert(0, str(script_dir))
|
|
|
|
from map_manager.engine import MapEngine
|
|
from map_manager.visualizer import MapVisualizer
|
|
|
|
|
|
def prompt():
|
|
return input('map-debug> ')
|
|
|
|
|
|
def parse_floats(args):
|
|
try:
|
|
return [float(x) for x in args]
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
class DebugApp:
|
|
def __init__(self):
|
|
self.service = 'osm'
|
|
self.cache_dir = 'debug_map_cache'
|
|
self.online = True
|
|
self.engine = MapEngine(service_name=self.service, cache_dir=self.cache_dir, enable_online=self.online)
|
|
self.visual = MapVisualizer(self.engine)
|
|
self.visual.set_click_callback(self.on_click)
|
|
self.last_image = None
|
|
|
|
def on_click(self, lat, lon):
|
|
print(f'Clicked callback: {lat}, {lon}')
|
|
|
|
def cmd_point(self, parts):
|
|
# point LAT LON ZOOM [RADIUS]
|
|
if len(parts) < 4:
|
|
print('Usage: point LAT LON ZOOM [RADIUS]')
|
|
return
|
|
vals = parse_floats(parts[1:])
|
|
if not vals or len(vals) < 3:
|
|
print('Invalid numeric arguments')
|
|
return
|
|
lat, lon, zoom = vals[0], vals[1], int(vals[2])
|
|
radius = int(vals[3]) if len(vals) >= 4 else 1
|
|
print(f'Fetching point {lat},{lon} @ zoom {zoom} radius {radius}')
|
|
img = self.engine.get_image_for_point(lat, lon, zoom, tiles_radius=radius)
|
|
if img is None:
|
|
print('No image returned.')
|
|
else:
|
|
self.last_image = img
|
|
self.visual.show_pil_image(img)
|
|
|
|
def cmd_area(self, parts):
|
|
# area W S E N [--zoom Z] [--maxsize PIX]
|
|
# minimal parser
|
|
if len(parts) < 5:
|
|
print('Usage: area W S E N [--zoom Z] [--maxsize PIX]')
|
|
return
|
|
vals = parse_floats(parts[1:5])
|
|
if not vals:
|
|
print('Invalid numeric bbox')
|
|
return
|
|
w, s, e, n = vals
|
|
zoom = None
|
|
maxsize = None
|
|
# parse optional flags
|
|
i = 5
|
|
while i < len(parts):
|
|
p = parts[i]
|
|
if p == '--zoom' and i + 1 < len(parts):
|
|
try:
|
|
zoom = int(parts[i+1])
|
|
except Exception:
|
|
pass
|
|
i += 2
|
|
elif p == '--maxsize' and i + 1 < len(parts):
|
|
try:
|
|
maxsize = int(parts[i+1])
|
|
except Exception:
|
|
pass
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
|
|
print(f'Fetching area bbox=({w},{s},{e},{n}) zoom={zoom} maxsize={maxsize}')
|
|
img = self.engine.get_image_for_area((w, s, e, n), zoom=zoom, max_size=maxsize)
|
|
if img is None:
|
|
print('No image returned.')
|
|
else:
|
|
self.last_image = img
|
|
self.visual.show_pil_image(img)
|
|
|
|
def cmd_save(self, parts):
|
|
if not self.last_image:
|
|
print('No image to save')
|
|
return
|
|
if len(parts) < 2:
|
|
print('Usage: save PATH')
|
|
return
|
|
path = parts[1]
|
|
try:
|
|
self.last_image.save(path)
|
|
print(f'Saved image to {path}')
|
|
except Exception as e:
|
|
print(f'Failed to save image: {e}')
|
|
|
|
def cmd_show(self, parts):
|
|
if not self.last_image:
|
|
print('No image to show')
|
|
return
|
|
self.visual.show_pil_image(self.last_image)
|
|
|
|
def cmd_toggle_online(self, parts):
|
|
self.online = not self.online
|
|
# Recreate engine with new flag
|
|
self.engine = MapEngine(service_name=self.service, cache_dir=self.cache_dir, enable_online=self.online)
|
|
self.visual = MapVisualizer(self.engine)
|
|
self.visual.set_click_callback(self.on_click)
|
|
print(f'Online fetching set to {self.online}')
|
|
|
|
def cmd_clear_cache(self, parts):
|
|
# clear entire service cache directory
|
|
try:
|
|
self.engine.tile_manager.clear_entire_service_cache()
|
|
print('Service cache cleared')
|
|
except Exception as e:
|
|
print(f'Failed to clear cache: {e}')
|
|
|
|
def repl(self):
|
|
print('Interactive map debug tool. Type "help" for commands.')
|
|
while True:
|
|
try:
|
|
line = prompt()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print('\nExiting')
|
|
return
|
|
if not line:
|
|
continue
|
|
parts = shlex.split(line)
|
|
cmd = parts[0].lower()
|
|
if cmd in ('exit', 'quit'):
|
|
print('Bye')
|
|
return
|
|
if cmd == 'help':
|
|
print('Commands: point, area, save, show, toggle_online, clear_cache, exit')
|
|
continue
|
|
if cmd == 'point':
|
|
self.cmd_point(parts)
|
|
continue
|
|
if cmd == 'area':
|
|
self.cmd_area(parts)
|
|
continue
|
|
if cmd == 'save':
|
|
self.cmd_save(parts)
|
|
continue
|
|
if cmd == 'show':
|
|
self.cmd_show(parts)
|
|
continue
|
|
if cmd == 'toggle_online':
|
|
self.cmd_toggle_online(parts)
|
|
continue
|
|
if cmd == 'clear_cache':
|
|
self.cmd_clear_cache(parts)
|
|
continue
|
|
print('Unknown command. Type help for commands.')
|
|
|
|
|
|
def main():
|
|
app = DebugApp()
|
|
app.repl()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|