"""
Per-action mouse handlers for GuiHydrology.
Mirrors the structure of ``_viewer_plugin_handlers.py``. Each function
extracted from ``GuiHydrology.On_Mouse_Right_Down`` / ``On_Mouse_Motion``
becomes an independent, testable callable registered via
``register_action()`` in ``GuiHydrology.__init__``.
Signature conventions
---------------------
* rdown / motion handlers: ``(viewer, MouseContext) -> None``
* ``viewer`` is the ``GuiHydrology`` instance — accessed as a plain object
(never imported at module level to avoid circular imports).
Dispatch tables
---------------
``HYDRO_RDOWN_HANDLERS`` — maps ActionKind → rdown handler
``HYDRO_MOTION_HANDLERS`` — maps ActionKind → motion handler
These tables are consumed by ``GuiHydrology.__init__`` when calling
``register_action()`` for each hydrology-specific action.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from ._action_kind import ActionKind
from ._viewer_plugin_handlers import MouseContext
from .PyTranslate import _
from .PyVertexvectors import zone, getIfromRGB
if TYPE_CHECKING:
from .PyGuiHydrology import GuiHydrology
# ---------------------------------------------------------------------------
# Internal helper
# ---------------------------------------------------------------------------
[docs]
def _watershed(viewer: 'GuiHydrology'):
"""Return the watershed or None (mirrors the @property in GuiHydrology)."""
if viewer.wolfparent is None:
return None
mc = viewer.wolfparent.mycatchment
if mc is None:
return None
return mc.charact_watrshd
# ---------------------------------------------------------------------------
# PICK_OUTLET
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_pick_outlet(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.wolfparent is None:
logging.warning(_('GuiHydrology: no wolfparent — cannot set outlet.'))
return
viewer.wolfparent.set_outlet(ctx.x, ctx.y)
viewer.Refresh()
# ---------------------------------------------------------------------------
# PICK_INTERIOR_POINT
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_pick_interior_point(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.wolfparent is None:
logging.warning(_('GuiHydrology: no wolfparent — cannot add interior point.'))
return
viewer.wolfparent.add_interior_point(ctx.x, ctx.y)
viewer.Refresh()
# ---------------------------------------------------------------------------
# REMOVE_FORCED_EXCHANGES
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_remove_forced_exchanges(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
exc = viewer.wolfparent.myexchanges if viewer.wolfparent is not None else None
if exc is None:
logging.warning(_('No forced exchanges to remove!'))
return
if exc.is_empty():
logging.warning(_('No forced exchanges to remove!'))
return
exc.remove_nearest_pair(ctx.x, ctx.y)
viewer.Refresh()
# ---------------------------------------------------------------------------
# FIND_UPSTREAM_WATERSHED / FIND_UPSTREAM_WATERSHED_LIMIT
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_find_upstream_watershed(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.active_array is None:
logging.warning(_('No active array — please select an active array first!'))
return
ws = _watershed(viewer)
if ws is None:
logging.warning(_('No watershed defined — please define a watershed first!'))
return
limit_to_sub = (viewer.action == ActionKind.FIND_UPSTREAM_WATERSHED_LIMIT)
starting_node = ws.get_node_from_xy(ctx.x, ctx.y)
up_vect = ws.get_vector_from_upstream_node(starting_node, limit_to_sub=limit_to_sub)
if up_vect is None:
logging.warning(_('No upstream watershed found!'))
return
def _style(vec):
vec.myprop.color = getIfromRGB((255, 0, 0))
vec.myprop.width = 3
vec.myprop.transparent = False
vec.myprop.alpha = 122
vec.myprop.filled = False
if viewer.active_array.Operations is not None:
newzone = zone(name=str(starting_node.sub))
viewer.active_array.Operations.show_structure_OpsVectors()
viewer.active_array.Operations.myzones.add_zone(newzone, forceparent=True)
newzone.add_vector(up_vect, forceparent=True)
_style(up_vect)
viewer.active_array.Operations.myzones.prep_listogl()
viewer.active_array.Operations.myzones.fill_structure()
viewer.Refresh()
else:
logging.warning(_('No operations frame in the active array!'))
# ---------------------------------------------------------------------------
# FIND_PATH_TO_OUTLET
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_find_path_to_outlet(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.active_array is None:
logging.warning(_('No active array — please select an active array first!'))
return
ws = _watershed(viewer)
if ws is None:
logging.warning(_('No watershed defined — please define a watershed first!'))
return
down_vect = ws.get_vector_from_xy_to_outlet(ctx.x, ctx.y)
if down_vect is None:
logging.warning(_('No downstream path found!'))
return
def _style(vec):
vec.myprop.color = getIfromRGB((255, 128, 192))
vec.myprop.width = 3
vec.myprop.transparent = False
vec.myprop.alpha = 122
vec.myprop.filled = False
if viewer.active_array.Operations is not None:
newzone = zone(name=down_vect.myname)
viewer.active_array.Operations.show_structure_OpsVectors()
viewer.active_array.Operations.myzones.add_zone(newzone, forceparent=True)
newzone.add_vector(down_vect, forceparent=True)
_style(down_vect)
viewer.active_array.Operations.myzones.prep_listogl()
viewer.active_array.Operations.myzones.fill_structure()
viewer.Refresh()
else:
logging.warning(_('No operations frame in the active array!'))
# ---------------------------------------------------------------------------
# SELECT_UPSTREAM_WATERSHED / SELECT_UPSTREAM_WATERSHED_LIMIT
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_select_upstream_watershed(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.active_array is None:
logging.warning(_('No active array — please select an active array first!'))
return
ws = _watershed(viewer)
if ws is None:
logging.warning(_('No watershed defined — please define a watershed first!'))
return
limit_to_sub = (viewer.action == ActionKind.SELECT_UPSTREAM_WATERSHED_LIMIT)
node = ws.get_node_from_xy(ctx.x, ctx.y)
xy = ws.get_xy_upstream_node(node, limit_to_sub=limit_to_sub)
viewer.active_array.SelectionData.set_selection_from_list_xy(xy)
viewer.Refresh()
# ---------------------------------------------------------------------------
# SELECT_UPSTREAM_RIVERS / SELECT_UPSTREAM_RIVERS_LIMIT
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_select_upstream_rivers(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.active_array is None:
logging.warning(_('No active array — please select an active array first!'))
return
ws = _watershed(viewer)
if ws is None:
logging.warning(_('No watershed defined — please define a watershed first!'))
return
limit_to_sub = (viewer.action == ActionKind.SELECT_UPSTREAM_RIVERS_LIMIT)
node = ws.get_node_from_xy(ctx.x, ctx.y)
xy = ws.get_xy_upstream_node(node, limit_to_sub=limit_to_sub, limit_to_river=True)
viewer.active_array.SelectionData.set_selection_from_list_xy(xy)
viewer.Refresh()
# ---------------------------------------------------------------------------
# SELECT_DOWNSTREAM_RIVERS
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_select_downstream_rivers(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
if viewer.active_array is None:
logging.warning(_('No active array — please select an active array first!'))
return
ws = _watershed(viewer)
if ws is None:
logging.warning(_('No watershed defined — please define a watershed first!'))
return
node = ws.get_node_from_xy(ctx.x, ctx.y)
xy = ws.get_xy_downstream_node(node)
viewer.active_array.SelectionData.set_selection_from_list_xy(xy)
viewer.Refresh()
# ---------------------------------------------------------------------------
# PICK_FORCED_EXCHANGES — rdown
# ---------------------------------------------------------------------------
[docs]
def _hydro_rdown_pick_forced_exchanges(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
from .PyDraw import draw_type # local import to avoid circulars
tmp_vec = viewer.wolfparent.myexchanges.temporary_vector
if tmp_vec.nbvertices == 0:
from .PyVertexvectors import wolfvertex as wv
tmp_vec.add_vertex(wv(ctx.x, ctx.y))
tmp_vec.add_vertex(wv(ctx.x, ctx.y))
elif tmp_vec.nbvertices == 2:
tmp_vec.myvertices[-1].x = ctx.x
tmp_vec.myvertices[-1].y = ctx.y
raw_data_elev = viewer.get_obj_from_id('Raw elevation [m]', drawing_type=draw_type.ARRAYS)
zup = zdown = float('nan')
if raw_data_elev is not None:
zup = raw_data_elev.get_value(tmp_vec.myvertices[0].x, tmp_vec.myvertices[0].y)
zdown = raw_data_elev.get_value(tmp_vec.myvertices[1].x, tmp_vec.myvertices[1].y)
txt = _('Do you want to add this forced exchange ?\n\n')
txt += _('Up node: {:.2f} m\n').format(zup)
txt += _('Down node: {:.2f} m').format(zdown)
if viewer._dialogs.ask_yes_no(txt, _('Forced exchange'), parent=viewer):
viewer.wolfparent.myexchanges.add_pair(
tmp_vec.myvertices[0].copy(),
tmp_vec.myvertices[1].copy(),
)
tmp_vec.reset()
# ---------------------------------------------------------------------------
# PICK_FORCED_EXCHANGES — motion
# ---------------------------------------------------------------------------
[docs]
def _hydro_motion_pick_forced_exchanges(viewer: 'GuiHydrology', ctx: MouseContext) -> None:
tmp_vec = viewer.wolfparent.myexchanges.temporary_vector
if tmp_vec.nbvertices == 2:
tmp_vec.myvertices[-1].x = ctx.x
tmp_vec.myvertices[-1].y = ctx.y
viewer.Refresh()
# ---------------------------------------------------------------------------
# Dispatch tables
# ---------------------------------------------------------------------------
[docs]
HYDRO_RDOWN_HANDLERS: dict = {
ActionKind.PICK_OUTLET: _hydro_rdown_pick_outlet,
ActionKind.PICK_INTERIOR_POINT: _hydro_rdown_pick_interior_point,
ActionKind.REMOVE_FORCED_EXCHANGES: _hydro_rdown_remove_forced_exchanges,
ActionKind.FIND_UPSTREAM_WATERSHED: _hydro_rdown_find_upstream_watershed,
ActionKind.FIND_UPSTREAM_WATERSHED_LIMIT: _hydro_rdown_find_upstream_watershed,
ActionKind.FIND_PATH_TO_OUTLET: _hydro_rdown_find_path_to_outlet,
ActionKind.SELECT_UPSTREAM_WATERSHED: _hydro_rdown_select_upstream_watershed,
ActionKind.SELECT_UPSTREAM_WATERSHED_LIMIT:_hydro_rdown_select_upstream_watershed,
ActionKind.SELECT_UPSTREAM_RIVERS: _hydro_rdown_select_upstream_rivers,
ActionKind.SELECT_UPSTREAM_RIVERS_LIMIT: _hydro_rdown_select_upstream_rivers,
ActionKind.SELECT_DOWNSTREAM_RIVERS: _hydro_rdown_select_downstream_rivers,
ActionKind.PICK_FORCED_EXCHANGES: _hydro_rdown_pick_forced_exchanges,
}
[docs]
HYDRO_MOTION_HANDLERS: dict = {
ActionKind.PICK_FORCED_EXCHANGES: _hydro_motion_pick_forced_exchanges,
}