256 lines
9.1 KiB
Python
256 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import dpg_map as dpgm
|
|
import dpg_map.renderer as renderer_module
|
|
from dpg_map.commands import CommandKind, MapCommand
|
|
from dpg_map.providers import TileProvider
|
|
from dpg_map.renderer import MapRenderer, drain_renderer_commands
|
|
from dpg_map.state import DirtyFlags, create_map_state
|
|
from dpg_map.tiles import TileID, TileResult, TileStatus, VisibleTile
|
|
|
|
|
|
class FakeDpg:
|
|
def __init__(self) -> None:
|
|
self.items: set[str | int] = set()
|
|
self.deleted: list[tuple[str | int, bool]] = []
|
|
self.drawn: list[tuple[str, str | int]] = []
|
|
self.images: list[tuple[Any, Any, str | int]] = []
|
|
|
|
def does_item_exist(self, tag: str | int) -> bool:
|
|
return tag in self.items
|
|
|
|
def add_draw_layer(self, *, parent: str | int, tag: str | int) -> None:
|
|
_ = parent
|
|
self.items.add(tag)
|
|
|
|
def delete_item(self, tag: str | int, *, children_only: bool = False) -> None:
|
|
self.deleted.append((tag, children_only))
|
|
|
|
def draw_rectangle(self, *args: Any, parent: str | int, **kwargs: Any) -> None:
|
|
_ = (args, kwargs)
|
|
self.drawn.append(("rectangle", parent))
|
|
|
|
def draw_image(self, *args: Any, parent: str | int, **kwargs: Any) -> None:
|
|
_ = (args, kwargs)
|
|
self.drawn.append(("image", parent))
|
|
self.images.append((args[1], args[2], parent))
|
|
|
|
def draw_text(self, *args: Any, parent: str | int, **kwargs: Any) -> None:
|
|
_ = (args, kwargs)
|
|
self.drawn.append(("text", parent))
|
|
|
|
def draw_circle(self, *args: Any, parent: str | int, **kwargs: Any) -> None:
|
|
_ = (args, kwargs)
|
|
self.drawn.append(("circle", parent))
|
|
|
|
def draw_polyline(self, *args: Any, parent: str | int, **kwargs: Any) -> None:
|
|
_ = (args, kwargs)
|
|
self.drawn.append(("polyline", parent))
|
|
|
|
|
|
class FrameCallbackDpg(FakeDpg):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.frame_count = 0
|
|
self.callbacks: dict[int, Any] = {}
|
|
|
|
def get_frame_count(self) -> int:
|
|
return self.frame_count
|
|
|
|
def set_frame_callback(self, frame: int, callback: Any) -> None:
|
|
self.callbacks[frame] = callback
|
|
|
|
|
|
def test_shared_frame_pump_renders_multiple_maps_from_one_frame_callback() -> None:
|
|
with renderer_module._scheduler_lock:
|
|
renderer_module._scheduled_renderers.clear()
|
|
renderer_module._scheduler_frame_scheduled = False
|
|
renderer_module._scheduler_dpg = None
|
|
|
|
state_a = create_map_state(tag="frame-pump-a")
|
|
state_b = create_map_state(tag="frame-pump-b")
|
|
fake = FrameCallbackDpg()
|
|
fake.items.update({state_a.drawlist_tag, state_b.drawlist_tag})
|
|
renderer_a = MapRenderer(state_a, fake)
|
|
renderer_b = MapRenderer(state_b, fake)
|
|
rendered: list[str] = []
|
|
renderer_a.render_frame = lambda: rendered.append("a") # type: ignore[method-assign]
|
|
renderer_b.render_frame = lambda: rendered.append("b") # type: ignore[method-assign]
|
|
|
|
renderer_a.schedule_next_frame()
|
|
renderer_b.schedule_next_frame()
|
|
assert list(fake.callbacks) == [1]
|
|
|
|
fake.frame_count = 1
|
|
fake.callbacks[1]()
|
|
|
|
assert rendered == ["a", "b"]
|
|
assert sorted(renderer_module._scheduled_renderers) == ["frame-pump-a", "frame-pump-b"]
|
|
|
|
with renderer_module._scheduler_lock:
|
|
renderer_module._scheduled_renderers.clear()
|
|
renderer_module._scheduler_frame_scheduled = False
|
|
renderer_module._scheduler_dpg = None
|
|
|
|
|
|
def test_renderer_command_drain_preserves_structural_order_and_coalesces() -> None:
|
|
state = create_map_state(tag="renderer-drain")
|
|
state.dirty = DirtyFlags.NONE
|
|
|
|
state.command_queue.put(MapCommand(CommandKind.ADD_OVERLAY, state.tag, {"tag": "a"}))
|
|
state.command_queue.put(MapCommand(CommandKind.SET_VIEW, state.tag, {"zoom": 3}))
|
|
state.command_queue.put(MapCommand(CommandKind.SET_VIEW, state.tag, {"zoom": 4}))
|
|
state.command_queue.put(MapCommand(CommandKind.UPDATE_OVERLAY, state.tag, {"tag": "a", "v": 1}))
|
|
state.command_queue.put(MapCommand(CommandKind.UPDATE_OVERLAY, state.tag, {"tag": "a", "v": 2}))
|
|
state.command_queue.put(MapCommand(CommandKind.DELETE_OVERLAY, state.tag, {"tag": "a"}))
|
|
|
|
commands = drain_renderer_commands(state)
|
|
|
|
assert [command.kind for command in commands] == [
|
|
CommandKind.ADD_OVERLAY,
|
|
CommandKind.SET_VIEW,
|
|
CommandKind.UPDATE_OVERLAY,
|
|
CommandKind.DELETE_OVERLAY,
|
|
]
|
|
assert commands[1].payload == {"zoom": 4}
|
|
assert commands[2].payload == {"tag": "a", "v": 2}
|
|
assert state.dirty & DirtyFlags.VIEW
|
|
assert state.dirty & DirtyFlags.OVERLAYS
|
|
|
|
|
|
def test_overlay_draw_clears_only_overlay_layer() -> None:
|
|
state = create_map_state(tag="overlay-draw", center=(47.0, 2.0), zoom=8)
|
|
dpgm.add_marker(
|
|
"vehicle",
|
|
lat=47.0,
|
|
lon=2.0,
|
|
show_label=True,
|
|
label="Vehicle",
|
|
map_tag="overlay-draw",
|
|
)
|
|
fake = FakeDpg()
|
|
fake.items.add(state.drawlist_tag)
|
|
renderer = MapRenderer(state, fake)
|
|
|
|
renderer._draw_tile_layer(
|
|
visible_tiles=[], width=400, height=300, attribution="Tiles", tile_size=256
|
|
)
|
|
fake.deleted.clear()
|
|
with state.lock:
|
|
overlays = tuple(state.overlays.values())
|
|
layers = {name: (layer.show, layer.z_index) for name, layer in state.layers.items()}
|
|
|
|
renderer._draw_overlay_layer(
|
|
overlays=overlays,
|
|
layers=layers,
|
|
center=state.center,
|
|
zoom=state.zoom,
|
|
width=400,
|
|
height=300,
|
|
tile_size=256,
|
|
)
|
|
|
|
assert fake.deleted == [("overlay-draw##layer-overlays", True)]
|
|
assert ("circle", "overlay-draw##layer-overlays") in fake.drawn
|
|
assert ("text", "overlay-draw##layer-overlays") in fake.drawn
|
|
|
|
|
|
def test_tile_draw_snaps_fractional_positions_to_integer_pixels() -> None:
|
|
state = create_map_state(tag="tile-snap", center=(47.0, 2.0), zoom=8)
|
|
tile_id = TileID("osm", 8, 129, 89)
|
|
state.tile_manager._result_queue.put(
|
|
TileResult(
|
|
tile_id,
|
|
generation=state.generation,
|
|
status=TileStatus.READY,
|
|
width=1,
|
|
height=1,
|
|
pixels=(1.0, 1.0, 1.0, 1.0),
|
|
source="disk",
|
|
)
|
|
)
|
|
state.tile_manager.drain_results(generation=state.generation, provider_name="osm")
|
|
state.tile_manager.set_texture_tag(tile_id, "tile-texture")
|
|
fake = FakeDpg()
|
|
fake.items.add(state.drawlist_tag)
|
|
renderer = MapRenderer(state, fake)
|
|
|
|
renderer._draw_tile_layer(
|
|
visible_tiles=[VisibleTile(tile_id, 10.4, 20.6)],
|
|
width=400,
|
|
height=300,
|
|
attribution="Tiles",
|
|
tile_size=256,
|
|
)
|
|
|
|
assert fake.images == [((10, 21), (266, 277), "tile-snap##layer-tiles")]
|
|
|
|
|
|
def test_overlay_update_drain_sets_only_overlay_dirty() -> None:
|
|
state = create_map_state(tag="overlay-dirty")
|
|
state.dirty = DirtyFlags.NONE
|
|
state.command_queue.put(MapCommand(CommandKind.UPDATE_OVERLAY, state.tag, {"tag": "a"}))
|
|
|
|
drain_renderer_commands(state)
|
|
|
|
assert state.dirty == DirtyFlags.OVERLAYS
|
|
|
|
|
|
def test_provider_switch_keeps_overlays_and_invalidates_tiles() -> None:
|
|
provider = TileProvider(
|
|
name="renderer-switch-provider",
|
|
url_template="https://tiles.example.test/{z}/{x}/{y}.png",
|
|
min_zoom=3,
|
|
max_zoom=4,
|
|
attribution="Example",
|
|
)
|
|
dpgm.register_provider(provider)
|
|
try:
|
|
state = create_map_state(tag="provider-switch", center=(47.0, 2.0), zoom=8)
|
|
dpgm.add_marker("vehicle", lat=47.0, lon=2.0, map_tag="provider-switch")
|
|
state.command_queue.drain()
|
|
state.dirty = DirtyFlags.NONE
|
|
tile_id = TileID("osm", 3, 1, 2)
|
|
state.tile_manager._result_queue.put(
|
|
TileResult(
|
|
tile_id,
|
|
generation=state.generation,
|
|
status=TileStatus.READY,
|
|
width=1,
|
|
height=1,
|
|
pixels=(1.0, 1.0, 1.0, 1.0),
|
|
source="disk",
|
|
)
|
|
)
|
|
state.tile_manager.drain_results(generation=state.generation, provider_name="osm")
|
|
state.tile_manager.set_texture_tag(tile_id, "old-texture")
|
|
|
|
dpgm.set_provider("renderer-switch-provider", map_tag="provider-switch")
|
|
drain_renderer_commands(state)
|
|
|
|
assert "vehicle" in state.overlays
|
|
assert state.center == (47.0, 2.0)
|
|
assert state.zoom == 4
|
|
assert state.generation == 1
|
|
assert state.provider.name == "renderer-switch-provider"
|
|
assert state.dirty & DirtyFlags.PROVIDER
|
|
assert state.dirty & DirtyFlags.TILES
|
|
assert state.dirty & DirtyFlags.OVERLAYS
|
|
assert state.tile_manager.get_ready_tile(tile_id) is None
|
|
assert state.tile_manager.take_texture_deletions() == ["old-texture"]
|
|
finally:
|
|
dpgm.unregister_provider("renderer-switch-provider")
|
|
|
|
|
|
def test_map_scoped_clear_disk_cache_command_keeps_dearpygui_out_of_caller_thread() -> None:
|
|
state = create_map_state(tag="clear-disk-command")
|
|
state.dirty = DirtyFlags.NONE
|
|
|
|
dpgm.clear_disk_cache(map_tag="clear-disk-command")
|
|
|
|
commands = state.command_queue.drain()
|
|
assert [command.kind for command in commands] == [CommandKind.CLEAR_DISK_CACHE]
|
|
assert state.dirty == DirtyFlags.TILES
|