333 lines
11 KiB
Python
333 lines
11 KiB
Python
"""Thread-safe state models and registries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, field
|
|
from enum import IntFlag, auto
|
|
from pathlib import Path
|
|
from threading import RLock
|
|
from uuid import uuid4
|
|
|
|
from .commands import MapCommandQueue
|
|
from .exceptions import InvalidProviderError, MapNotFoundError
|
|
from .overlays import LayerState, Overlay
|
|
from .providers import TileProvider, get_default_provider, get_provider
|
|
from .tiles import TileManager
|
|
from .types import LatLon, Tag
|
|
|
|
|
|
class DirtyFlags(IntFlag):
|
|
"""Reasons the GUI renderer needs to refresh part of a map."""
|
|
|
|
NONE = 0
|
|
VIEW = auto()
|
|
TILES = auto()
|
|
OVERLAYS = auto()
|
|
SIZE = auto()
|
|
PROVIDER = auto()
|
|
FULL = VIEW | TILES | OVERLAYS | SIZE | PROVIDER
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DpgMapConfig:
|
|
"""Global package configuration."""
|
|
|
|
user_agent: str | None = None
|
|
cache_dir: Path | None = None
|
|
default_provider: str | TileProvider = "osm"
|
|
memory_cache_max_tiles: int = 512
|
|
disk_cache_max_bytes: int | None = 2_000_000_000
|
|
prefetch_margin_tiles: int = 1
|
|
tile_worker_count: int = 4
|
|
overlay_update_policy: str = "coalesce"
|
|
debug: bool = False
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class InteractionState:
|
|
"""Logical mouse interaction state."""
|
|
|
|
active_drag: bool = False
|
|
last_mouse_position: tuple[float, float] | None = None
|
|
|
|
|
|
def default_layers() -> dict[str, LayerState]:
|
|
"""Return the default logical map layers."""
|
|
|
|
layers = [
|
|
LayerState("background", z_index=0),
|
|
LayerState("tiles", z_index=10),
|
|
LayerState("default", z_index=50),
|
|
LayerState("markers", z_index=60),
|
|
LayerState("lines", z_index=70),
|
|
LayerState("trajectories", z_index=80),
|
|
LayerState("attribution", z_index=100),
|
|
]
|
|
return {layer.name: layer for layer in layers}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class MapState:
|
|
"""Thread-safe logical state for one map widget."""
|
|
|
|
tag: Tag
|
|
child_window_tag: Tag
|
|
drawlist_tag: Tag
|
|
texture_registry_tag: Tag
|
|
handler_registry_tag: Tag
|
|
requested_width: int = 0
|
|
requested_height: int = 0
|
|
requested_autosize_x: bool = False
|
|
requested_autosize_y: bool = False
|
|
measured_width: int = 0
|
|
measured_height: int = 0
|
|
last_nonzero_width: int = 0
|
|
last_nonzero_height: int = 0
|
|
is_visible: bool = False
|
|
center: LatLon = (0.0, 0.0)
|
|
zoom: int = 2
|
|
min_zoom: int = 0
|
|
max_zoom: int = 19
|
|
provider: TileProvider = field(default_factory=get_default_provider)
|
|
overlays: dict[Tag, Overlay] = field(default_factory=dict)
|
|
layers: dict[str, LayerState] = field(default_factory=default_layers)
|
|
command_queue: MapCommandQueue = field(default_factory=MapCommandQueue)
|
|
tile_manager: TileManager = field(default_factory=TileManager)
|
|
renderer: object | None = None
|
|
interaction: InteractionState = field(default_factory=InteractionState)
|
|
lock: RLock = field(default_factory=RLock)
|
|
dirty: DirtyFlags = DirtyFlags.FULL
|
|
frame_scheduled: bool = False
|
|
generation: int = 0
|
|
cache_dir: Path | None = None
|
|
user_agent: str | None = None
|
|
|
|
|
|
_config = DpgMapConfig()
|
|
_config_lock = RLock()
|
|
_maps: dict[Tag, MapState] = {}
|
|
_maps_lock = RLock()
|
|
_current_map_stack: list[Tag] = []
|
|
_current_map_lock = RLock()
|
|
|
|
|
|
def _resolve_provider(provider: str | TileProvider | None) -> TileProvider:
|
|
if provider is None:
|
|
with _config_lock:
|
|
provider = _config.default_provider
|
|
if isinstance(provider, TileProvider):
|
|
return provider
|
|
if isinstance(provider, str):
|
|
return get_provider(provider)
|
|
raise InvalidProviderError("provider must be a provider name or TileProvider")
|
|
|
|
|
|
def configure_state(
|
|
*,
|
|
user_agent: str | None = None,
|
|
cache_dir: str | Path | None = None,
|
|
default_provider: str | TileProvider = "osm",
|
|
memory_cache_max_tiles: int = 512,
|
|
disk_cache_max_bytes: int | None = 2_000_000_000,
|
|
prefetch_margin_tiles: int = 1,
|
|
tile_worker_count: int = 4,
|
|
overlay_update_policy: str = "coalesce",
|
|
debug: bool = False,
|
|
) -> None:
|
|
"""Replace global dpg-map configuration."""
|
|
|
|
if memory_cache_max_tiles < 0:
|
|
raise ValueError("memory_cache_max_tiles must be >= 0")
|
|
if disk_cache_max_bytes is not None and disk_cache_max_bytes < 0:
|
|
raise ValueError("disk_cache_max_bytes must be >= 0 or None")
|
|
if prefetch_margin_tiles < 0:
|
|
raise ValueError("prefetch_margin_tiles must be >= 0")
|
|
if tile_worker_count < 1:
|
|
raise ValueError("tile_worker_count must be >= 1")
|
|
if overlay_update_policy != "coalesce":
|
|
raise ValueError('overlay_update_policy must be "coalesce"')
|
|
|
|
resolved_cache_dir = Path(cache_dir).expanduser() if cache_dir is not None else None
|
|
_resolve_provider(default_provider)
|
|
with _config_lock:
|
|
_config.user_agent = user_agent
|
|
_config.cache_dir = resolved_cache_dir
|
|
_config.default_provider = default_provider
|
|
_config.memory_cache_max_tiles = memory_cache_max_tiles
|
|
_config.disk_cache_max_bytes = disk_cache_max_bytes
|
|
_config.prefetch_margin_tiles = prefetch_margin_tiles
|
|
_config.tile_worker_count = tile_worker_count
|
|
_config.overlay_update_policy = overlay_update_policy
|
|
_config.debug = debug
|
|
|
|
|
|
def get_config() -> DpgMapConfig:
|
|
"""Return a copy of current global configuration."""
|
|
|
|
with _config_lock:
|
|
return DpgMapConfig(
|
|
user_agent=_config.user_agent,
|
|
cache_dir=_config.cache_dir,
|
|
default_provider=_config.default_provider,
|
|
memory_cache_max_tiles=_config.memory_cache_max_tiles,
|
|
disk_cache_max_bytes=_config.disk_cache_max_bytes,
|
|
prefetch_margin_tiles=_config.prefetch_margin_tiles,
|
|
tile_worker_count=_config.tile_worker_count,
|
|
overlay_update_policy=_config.overlay_update_policy,
|
|
debug=_config.debug,
|
|
)
|
|
|
|
|
|
def create_map_state(
|
|
*,
|
|
tag: Tag | None = None,
|
|
center: LatLon = (0.0, 0.0),
|
|
zoom: int = 2,
|
|
min_zoom: int | None = None,
|
|
max_zoom: int | None = None,
|
|
width: int = 0,
|
|
height: int = 0,
|
|
autosize_x: bool = False,
|
|
autosize_y: bool = False,
|
|
provider: str | TileProvider | None = None,
|
|
cache_dir: str | Path | None = None,
|
|
user_agent: str | None = None,
|
|
) -> MapState:
|
|
"""Create and register a logical map state."""
|
|
|
|
map_tag = tag if tag is not None else f"dpg_map_{uuid4().hex}"
|
|
provider_obj = _resolve_provider(provider)
|
|
min_zoom_value = provider_obj.min_zoom if min_zoom is None else min_zoom
|
|
max_zoom_value = provider_obj.max_zoom if max_zoom is None else max_zoom
|
|
if min_zoom_value < provider_obj.min_zoom:
|
|
min_zoom_value = provider_obj.min_zoom
|
|
if max_zoom_value > provider_obj.max_zoom:
|
|
max_zoom_value = provider_obj.max_zoom
|
|
if max_zoom_value < min_zoom_value:
|
|
raise ValueError("max_zoom must be >= min_zoom")
|
|
|
|
zoom_value = max(min_zoom_value, min(max_zoom_value, int(zoom)))
|
|
config = get_config()
|
|
resolved_cache_dir = Path(cache_dir).expanduser() if cache_dir is not None else config.cache_dir
|
|
state = MapState(
|
|
tag=map_tag,
|
|
child_window_tag=f"{map_tag}##child",
|
|
drawlist_tag=f"{map_tag}##drawlist",
|
|
texture_registry_tag=f"{map_tag}##textures",
|
|
handler_registry_tag=f"{map_tag}##handlers",
|
|
requested_width=width,
|
|
requested_height=height,
|
|
requested_autosize_x=autosize_x,
|
|
requested_autosize_y=autosize_y,
|
|
center=(float(center[0]), float(center[1])),
|
|
zoom=zoom_value,
|
|
min_zoom=min_zoom_value,
|
|
max_zoom=max_zoom_value,
|
|
provider=provider_obj,
|
|
cache_dir=resolved_cache_dir,
|
|
user_agent=user_agent if user_agent is not None else config.user_agent,
|
|
)
|
|
state.tile_manager = TileManager(
|
|
memory_cache_max_tiles=config.memory_cache_max_tiles,
|
|
disk_cache_max_bytes=config.disk_cache_max_bytes,
|
|
worker_count=config.tile_worker_count,
|
|
user_agent=state.user_agent,
|
|
)
|
|
with _maps_lock:
|
|
_maps[map_tag] = state
|
|
return state
|
|
|
|
|
|
def register_map_state(state: MapState) -> None:
|
|
"""Register a prebuilt map state."""
|
|
|
|
with _maps_lock:
|
|
_maps[state.tag] = state
|
|
|
|
|
|
def unregister_map_state(tag: Tag) -> None:
|
|
"""Remove a map state from the registry."""
|
|
|
|
with _maps_lock:
|
|
_maps.pop(tag, None)
|
|
|
|
|
|
def get_map_state(map_tag: Tag | None = None) -> MapState:
|
|
"""Resolve a map by explicit tag or current map context."""
|
|
|
|
resolved_tag = resolve_map_tag(map_tag)
|
|
with _maps_lock:
|
|
try:
|
|
return _maps[resolved_tag]
|
|
except KeyError as exc:
|
|
raise MapNotFoundError(f"map not registered: {resolved_tag}") from exc
|
|
|
|
|
|
def resolve_map_tag(map_tag: Tag | None = None) -> Tag:
|
|
"""Resolve an explicit tag or the current context map tag."""
|
|
|
|
if map_tag is not None:
|
|
return map_tag
|
|
with _current_map_lock:
|
|
if _current_map_stack:
|
|
return _current_map_stack[-1]
|
|
raise MapNotFoundError("map_tag is required outside a map_widget context")
|
|
|
|
|
|
def list_map_states() -> list[MapState]:
|
|
"""Return registered map states as a snapshot."""
|
|
|
|
with _maps_lock:
|
|
return list(_maps.values())
|
|
|
|
|
|
def find_map_for_overlay(tag: Tag, map_tag: Tag | None = None) -> MapState:
|
|
"""Find the map containing an overlay, optionally scoped by map tag."""
|
|
|
|
if map_tag is not None:
|
|
return get_map_state(map_tag)
|
|
|
|
with _current_map_lock:
|
|
if _current_map_stack:
|
|
current = get_map_state(_current_map_stack[-1])
|
|
with current.lock:
|
|
if tag in current.overlays:
|
|
return current
|
|
|
|
matches: list[MapState] = []
|
|
with _maps_lock:
|
|
states = list(_maps.values())
|
|
for state in states:
|
|
with state.lock:
|
|
if tag in state.overlays:
|
|
matches.append(state)
|
|
if len(matches) == 1:
|
|
return matches[0]
|
|
if not matches:
|
|
raise MapNotFoundError(f"no map contains overlay: {tag}")
|
|
raise MapNotFoundError(f"overlay tag is ambiguous across maps: {tag}")
|
|
|
|
|
|
@contextmanager
|
|
def current_map_context(map_tag: Tag) -> Iterator[None]:
|
|
"""Push a current map tag for context-style overlay creation."""
|
|
|
|
with _current_map_lock:
|
|
_current_map_stack.append(map_tag)
|
|
try:
|
|
yield
|
|
finally:
|
|
with _current_map_lock:
|
|
if _current_map_stack and _current_map_stack[-1] == map_tag:
|
|
_current_map_stack.pop()
|
|
elif map_tag in _current_map_stack:
|
|
_current_map_stack.remove(map_tag)
|
|
|
|
|
|
def mark_dirty(state: MapState, flags: DirtyFlags) -> None:
|
|
"""Mark a map dirty while holding or acquiring its state lock."""
|
|
|
|
state.dirty |= flags
|