"""DTM Wallonie 1 m/2 m — built-in companion plugin.
Loads the Wallonia 1 m DTM (LIDAXE) raster for the viewer's current zoom
extent and adds it to the active WolfMapViewer as a *WolfArray*.
Workflow
--------
1. On first use the plugin downloads the tile-index shapefile via
:func:`~wolfhece.pydownloader.toys_dtm_wallonia_1m`.
2. When "Charger depuis le zoom actuel" is triggered it:
- reads the current view bounds from the viewer,
- delegates extraction to :class:`~wolfhece.services.dtm_wallonia.DtmWalloniaService`,
- inserts the resulting :class:`~wolfhece.wolf_array.WolfArray` in the
viewer tree.
"""
from __future__ import annotations
from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING
import numpy as np
from wolfhece.plugins.abc import AbstractCompanionModel, AbstractUICompanion, MenuItem, SEPARATOR
from wolfhece.PyTranslate import _
from wolfhece.services import dtm_wallonia
from wolfhece._viewer_plugin_handlers import MouseContext
if TYPE_CHECKING:
from wolfhece.wolf_array import WolfArray
from wolfhece.pyvertexvectors import vector
[docs]
_logger = logging.getLogger(__name__)
# Backward-compatible alias used by existing plugin tests.
[docs]
_DtmWalloniaTiles = dtm_wallonia._DtmWalloniaTiles
# ---------------------------------------------------------------------------
# Business model (domain state)
# ---------------------------------------------------------------------------
@dataclass
[docs]
class DtmWalloniaModel(AbstractCompanionModel):
"""Pure plugin state independent from viewer/wx/OpenGL APIs.
This model intentionally stores only business/session state so that the UI
companion remains focused on event wiring and viewer integration.
"""
[docs]
last_layer_id: str | None = None
[docs]
last_bounds: tuple[float, float, float, float] | None = None
[docs]
_service: 'dtm_wallonia.DtmWalloniaService | None' = None
[docs]
def reset(self) -> None:
self.load_count = 0
self.last_layer_id = None
self.last_bounds = None
self._service = None
[docs]
def next_layer_id(self) -> str:
self.load_count += 1
self.last_layer_id = f'DTM_Wallonia_{self.load_count}'
return self.last_layer_id
@staticmethod
[docs]
def is_valid_extent(xmin: float, xmax: float, ymin: float, ymax: float) -> bool:
return xmin != xmax and ymin != ymax
@property
[docs]
def has_service(self) -> bool:
"""Whether the cached DTM service is already initialized."""
return self._service is not None
[docs]
def ensure_service(self) -> 'dtm_wallonia.DtmWalloniaService':
"""Return the cached DTM service, creating it if needed."""
if self._service is None:
self._service = dtm_wallonia.DtmWalloniaService()
return self._service
[docs]
def force_download_index(self) -> str:
"""Force (re-)download of the tile-index shapefile."""
return self.ensure_service().force_download_index()
[docs]
def get_dtm_1m(self, bounds_vec: 'vector', force: bool = False) -> 'WolfArray':
"""Return 1m DTM for the given bounds vector."""
return self.ensure_service().get_dtm_1m(bounds_vec, force=force)
[docs]
def get_dtm_2m(self, bounds_vec: 'vector', force: bool = False) -> 'WolfArray':
"""Return 2m DTM for the given bounds vector."""
return self.ensure_service().get_dtm_2m(bounds_vec, force=force)
[docs]
def get_dtm_1m_xx_yy(
self,
xmin: float,
xmax: float,
ymin: float,
ymax: float,
force: bool = False,
) -> 'WolfArray':
"""Return 1m DTM for raw bounds."""
return self.ensure_service().get_dtm_1m_xx_yy(
xmin=xmin,
xmax=xmax,
ymin=ymin,
ymax=ymax,
force=force,
)
[docs]
def get_dtm_2m_xx_yy(
self,
xmin: float,
xmax: float,
ymin: float,
ymax: float,
force: bool = False,
) -> 'WolfArray':
"""Return 2m DTM for raw bounds."""
return self.ensure_service().get_dtm_2m_xx_yy(
xmin=xmin,
xmax=xmax,
ymin=ymin,
ymax=ymax,
force=force,
)
[docs]
def load_from_extent(
self,
xmin: float,
xmax: float,
ymin: float,
ymax: float,
force: bool = False,
) -> tuple['WolfArray', str]:
"""Execute the domain workflow for one extent.
1. Validate extent,
2. Build closed bounds vector,
3. Fetch DTM,
4. Allocate next layer id.
"""
from wolfhece.PyVertexvectors import vector
if not self.is_valid_extent(xmin, xmax, ymin, ymax):
raise ValueError('invalid_extent')
self.last_bounds = (xmin, xmax, ymin, ymax)
bounds_vec = vector()
bounds_vec.add_vertices_from_array(np.array([
[xmin, ymin],
[xmax, ymin],
[xmax, ymax],
[xmin, ymax],
]))
bounds_vec.force_to_close()
dtm_array = self.get_dtm_1m(bounds_vec, force=force)
layer_id = self.next_layer_id()
return dtm_array, layer_id
# ---------------------------------------------------------------------------
# Companion
# ---------------------------------------------------------------------------
[docs]
class DtmWallonia1mCompanion(AbstractUICompanion):
"""Companion plugin that loads Wallonia DTM for the current zoom.
Menu: **DTM Wallonie 1m**
* *Load from current zoom* — adds a 1 m DTM layer for the current view.
* Public helpers also expose 2 m DTM (mean-rebinned from 1 m).
* *Télécharger l'index des tuiles* — (re-)downloads only the tile-index
shapefile, useful for an initial setup or refresh.
"""
[docs]
def create_model(self) -> DtmWalloniaModel:
return DtmWalloniaModel()
@property
[docs]
def model_state(self) -> DtmWalloniaModel:
return self.model # type: ignore[return-value]
[docs]
def start(self) -> None:
"""Programmatic activation — load DTM for the current view."""
self._do_load_from_zoom()
# ------------------------------------------------------------------
# Menu-item handlers
# ------------------------------------------------------------------
[docs]
def _on_load_from_zoom(self, _ctx: MouseContext) -> None:
self.start()
[docs]
def _on_load_from_zoom_force(self, _ctx: MouseContext) -> None:
self._do_load_from_zoom(force=True)
[docs]
def _on_show_index(self, _ctx: MouseContext) -> None:
self._show_index_in_viewer()
[docs]
def _on_download_index(self, _ctx: MouseContext) -> None:
self._do_download_index()
# ------------------------------------------------------------------
# Implementation
# ------------------------------------------------------------------
[docs]
def _show_index_in_viewer(self) -> None:
"""Add the tile-index grid to the viewer, using display lists for performance.
Idempotent — does nothing if the layer is already present.
Reuses the already-loaded :attr:`_DtmWalloniaTiles._tiles` object
(which already has ``VectorOGLRenderer.LIST`` set) instead of
re-reading the shapefile.
"""
from wolfhece.PyDraw import draw_type
_INDEX_ID = 'DTM_Wallonia_1m_index'
if _INDEX_ID in self.proxy._viewer.get_list_keys(draw_type.VECTORS, checked_state=None):
return
if not self.model_state.has_service:
self.proxy.set_status(_('Downloading tile index…'))
tiles = self.model_state.ensure_service().get_tiles_manager()
self.proxy._viewer.add_object('vector', newobj=tiles._tiles, ToCheck=True, id=_INDEX_ID)
self.proxy.set_status(_('Tile index added to viewer (display-list renderer).'))
[docs]
def _do_download_index(self) -> None:
"""Force (re-)download of the tile-index shapefile."""
self.proxy.set_status(_('Downloading tile index…'))
idx_path = self.model_state.force_download_index()
self.proxy.set_status(_('Tile index downloaded: {path}').format(path=idx_path))
[docs]
def get_dtm_1m(self, bounds_vec: 'vector', force: bool = False) -> 'WolfArray':
"""Return a :class:`~wolfhece.wolf_array.WolfArray` cropped to *bounds_vec*.
Downloads any missing tiles that intersect the polygon before
assembling the result.
:param bounds_vec: A closed :class:`~wolfhece.PyVertexvectors.vector`
polygon that defines the region of interest.
:param force: If True, bypass the >100 tiles safeguard in
:meth:`_DtmWalloniaTiles.extract`.
:returns: WolfArray covering the bounding box of *bounds_vec*.
:raises FileNotFoundError: When no tile can be downloaded for the
requested area.
"""
try:
if not self.model_state.has_service:
self.proxy.set_status(_('Downloading tile index…'))
return self.model_state.get_dtm_1m(bounds_vec, force=force)
except Exception:
_logger.exception('DtmWallonia1m: extract failed (force=%s)', force)
raise
[docs]
def get_dtm_2m(self, bounds_vec: 'vector', force: bool = False) -> 'WolfArray':
"""Return a 2 m :class:`~wolfhece.wolf_array.WolfArray` by mean-rebinning the 1 m raster."""
try:
if not self.model_state.has_service:
self.proxy.set_status(_('Downloading tile index…'))
return self.model_state.get_dtm_2m(bounds_vec, force=force)
except Exception:
_logger.exception('DtmWallonia1m: 2m extract failed (force=%s)', force)
raise
[docs]
def get_dtm_1m_xx_yy(self, xmin: float, xmax: float, ymin: float, ymax: float, force: bool = False) -> 'WolfArray':
"""Convenience wrapper around :meth:`get_dtm_1m` that takes raw bounds.
:param xmin: Minimum X coordinate of the bounding box.
:param xmax: Maximum X coordinate of the bounding box.
:param ymin: Minimum Y coordinate of the bounding box.
:param ymax: Maximum Y coordinate of the bounding box.
:param force: If True, bypass the >100 tiles safeguard in :meth:`_DtmWalloniaTiles.extract`.
:returns: WolfArray covering the specified bounding box.
:raises FileNotFoundError: When no tile can be downloaded for the requested area.
"""
try:
if not self.model_state.has_service:
self.proxy.set_status(_('Downloading tile index…'))
return self.model_state.get_dtm_1m_xx_yy(
xmin=xmin,
xmax=xmax,
ymin=ymin,
ymax=ymax,
force=force,
)
except Exception:
_logger.exception(
'DtmWallonia1m: get_dtm_xx_yy failed for bounds '
'(xmin=%s, xmax=%s, ymin=%s, ymax=%s, force=%s)',
xmin,
xmax,
ymin,
ymax,
force,
)
raise
[docs]
def get_dtm_2m_xx_yy(self, xmin: float, xmax: float, ymin: float, ymax: float, force: bool = False) -> 'WolfArray':
"""Convenience wrapper around :meth:`get_dtm_2m` that takes raw bounds."""
try:
if not self.model_state.has_service:
self.proxy.set_status(_('Downloading tile index…'))
return self.model_state.get_dtm_2m_xx_yy(
xmin=xmin,
xmax=xmax,
ymin=ymin,
ymax=ymax,
force=force,
)
except Exception:
_logger.exception(
'DtmWallonia1m: get_dtm_2m_xx_yy failed for bounds '
'(xmin=%s, xmax=%s, ymin=%s, ymax=%s, force=%s)',
xmin,
xmax,
ymin,
ymax,
force,
)
raise
[docs]
def _do_load_from_zoom(self, force: bool = False) -> None:
"""Load DTM for the current view extent and add it to the viewer."""
zoom = self.proxy._viewer.get_current_zoom()
xmin = zoom['xmin']
xmax = zoom['xmax']
ymin = zoom['ymin']
ymax = zoom['ymax']
if force:
self.proxy.set_status(_('Loading Wallonia 1m DTM (force mode)…'))
else:
self.proxy.set_status(_('Loading Wallonia 1m DTM…'))
try:
dtm_array, layer_id = self.model_state.load_from_extent(
xmin=xmin,
xmax=xmax,
ymin=ymin,
ymax=ymax,
force=force,
)
except ValueError:
self.proxy.set_status(_('Invalid current extent — zoom in first.'))
return
except Exception as exc:
_logger.error('DtmWallonia1m: load failed: %s', exc)
self.proxy.set_status(_('Error loading DTM: {err}').format(err=exc))
return
self.proxy._viewer.add_object('array', newobj=dtm_array, ToCheck=True, id=layer_id)
self.proxy.set_status(_('Wallonia 1m DTM loaded ({id}).').format(id=layer_id))