PlotKPC.py

# PlotKPC.py -- Pamhyr
# Copyright (C) 2023-2024  INRAE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

# -*- coding: utf-8 -*-

import logging

from functools import reduce

from tools import timer
from View.Tools.PamhyrPlot import PamhyrPlot

from PyQt5.QtCore import (
    QCoreApplication
)

logger = logging.getLogger()

_translate = QCoreApplication.translate


class PlotKPC(PamhyrPlot):
    def __init__(self, canvas=None, trad=None, toolbar=None,
                 results=None, reach_id=0, profile_id=0,
                 parent=None):
        super(PlotKPC, self).__init__(
            canvas=canvas,
            trad=trad,
            data=results,
            toolbar=toolbar,
            parent=parent
        )

        self._timestamps = results.get("timestamps")
        self._current_timestamp = max(self._timestamps)
        self._current_reach_id = reach_id
        self._current_profile_id = profile_id

        self.label_x = _translate("Results", "KP (m)")
        self.label_y = _translate("Results", "Elevation (m)")

        self.label_bottom = _translate("Results", "River bottom")
        self.label_water = _translate("Results", "Water elevation")
        self.label_water_max = _translate("Results", "Max water elevation")

        self._isometric_axis = False

    @property
    def results(self):
        return self.data

    @results.setter
    def results(self, results):
        self.data = results
        self._current_timestamp = max(results.get("timestamps"))

    @timer
    def draw(self, highlight=None):
        self.init_axes()

        if self.results is None:
            return

        reach = self.results.river.reach(self._current_reach_id)

        self.draw_bottom(reach)
        self.draw_water_elevation(reach)
        self.draw_water_elevation_max(reach)
        self.draw_water_elevation_overflow(reach)
        self.draw_current(reach)

        # self.enable_legend()

        self.idle()
        self._init = True

    def draw_bottom(self, reach):
        if reach.has_sediment():
            self.draw_bottom_with_bedload(reach)
        else:
            self.draw_bottom_geometry(reach)

    def draw_bottom_with_bedload(self, reach):
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()
        z_max = reach.geometry.get_z_max()

        initial_sl = self.sl_compute_initial(reach)
        current_sl = self.sl_compute_current(reach)

        max_sl_num = reduce(
            lambda acc, sl: max(acc, len(sl)),
            current_sl, 0
        )

        sl_init, sl = self.sl_completed_layers(
            initial_sl, current_sl, max_sl_num
        )

        z_sl = self.sl_apply_z_min_to_initial(sl_init, z_min)
        d_sl = self.sl_compute_diff(sl_init, sl)

        final_z_sl = self.sl_apply_diff(reach, z_sl, d_sl)
        final_z_sl = list(reversed(final_z_sl))

        self.line_kp_sl = []
        for i, z in enumerate(final_z_sl):
            self.line_kp_sl.append(None)
            self.line_kp_sl[i], = self.canvas.axes.plot(
                kp, z,
                linestyle="solid" if i == len(final_z_sl) - 1 else "--",
                lw=1.,
                color=(
                    self.color_plot_river_bottom if i == len(final_z_sl) - 1
                    else self.colors[i]
                )
            )

        self._initial_sl = initial_sl
        self._river_bottom = final_z_sl[-1]

    def sl_compute_initial(self, reach):
        """
        Get SL list for profile p at initial time (initial data)
        """
        return list(
            map(
                lambda p: p.get_ts_key(min(self._timestamps), "sl")[0],
                reach.profiles
            )
        )

    def sl_compute_current(self, reach):
        """
        Get SL list for profile p at current time
        """
        return list(
            map(
                lambda p: p.get_ts_key(self._current_timestamp, "sl")[0],
                reach.profiles
            )
        )

    def sl_completed_layers(self, initial_sl, current_sl, max_sl_num):
        sl = []
        sl_init = []
        for i in range(max_sl_num):
            cur = []
            cur_init = []
            for current, initial in zip(current_sl, initial_sl):
                if i < len(initial_sl):
                    cur.append(current[i][0])
                    cur_init.append(initial[i][0])
                else:
                    cur.append(0)
                    cur_init.append(0)
            sl.append(cur)
            sl_init.append(cur_init)

        return sl_init, sl

    def sl_apply_z_min_to_initial(self, sl_init, z_min):
        """
        Compute sediment layer from initial data in function to profile
        z_min
        """
        return reduce(
            lambda acc, v: acc + [
                list(
                    map(
                        lambda x, y: y - x,
                        v, acc[-1]
                    )
                )
            ],
            sl_init, [z_min]
        )

    def sl_compute_diff(self, sl_init, sl):
        return list(
            map(
                lambda ln0, lni: list(
                    map(
                        lambda z0, zi: z0 - zi,
                        ln0, lni
                    )
                ),
                sl_init, sl
            )
        )

    def sl_apply_diff(self, reach, z_sl, d_sl):
        f = list(map(lambda p: 0, reach.profiles))

        return list(
            map(
                lambda z, d: list(
                    map(
                        lambda zn, dn: zn - dn,
                        z, d
                    )
                ),
                z_sl, d_sl + [f]  # HACK: Add dummy data for last layer
            )
        )

    def draw_bottom_geometry(self, reach):
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()
        z_max = reach.geometry.get_z_max()

        self.line_kp_zmin = self.canvas.axes.plot(
            kp, z_min,
            color=self.color_plot_river_bottom,
            lw=1.
        )

        self._river_bottom = z_min

    def draw_water_elevation(self, reach):
        if len(reach.geometry.profiles) != 0:
            kp = reach.geometry.get_kp()
            z_min = reach.geometry.get_z_min()

            water_z = list(
                map(
                    lambda p: p.get_ts_key(
                        self._current_timestamp, "Z"
                    ),
                    reach.profiles
                )
            )

            self.water = self.canvas.axes.plot(
                kp, water_z,
                lw=1., color=self.color_plot_river_water,
            )

            self.water_fill = self.canvas.axes.fill_between(
                kp, self._river_bottom, water_z,
                color=self.color_plot_river_water_zone,
                alpha=0.7,
                interpolate=True
            )

    def draw_water_elevation_max(self, reach):
        if len(reach.geometry.profiles) != 0:
            kp = reach.geometry.get_kp()
            z_min = reach.geometry.get_z_min()

            water_z = list(
                map(
                    lambda p: max(p.get_key("Z")),
                    reach.profiles
                )
            )

            self.canvas.axes.plot(
                kp, water_z, lw=1.,
                color=self.color_plot_river_water,
                linestyle='dotted',
            )

    def draw_current(self, reach):
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()
        z_max = reach.geometry.get_z_max()

        self.profile, = self.canvas.axes.plot(
            [
                kp[self._current_profile_id],
                kp[self._current_profile_id]
            ],
            [
                z_max[self._current_profile_id],
                z_min[self._current_profile_id]
            ],
            color=self.color_plot,
            lw=1.
        )

    def draw_water_elevation_overflow(self, reach):
        overflow = []

        for profile in reach.profiles:
            z_max = max(profile.get_key("Z"))
            z_max_ts = 0
            for ts in self._timestamps:
                z = profile.get_ts_key(ts, "Z")
                if z == z_max:
                    z_max_ts = ts
                    break

            pt_left, pt_right = profile.get_ts_key(z_max_ts, "water_limits")

            if self.is_overflow_point(profile, pt_left):
                overflow.append((profile, z_max))
            elif self.is_overflow_point(profile, pt_right):
                overflow.append((profile, z_max))

        for profile, z in overflow:
            self.canvas.axes.plot(
                profile.kp, z,
                lw=1.,
                color=self.color_plot,
                markersize=3,
                marker='x'
            )

    def is_overflow_point(self, profile, point):
        left_limit = profile.geometry.point(0)
        right_limit = profile.geometry.point(
            profile.geometry.number_points - 1
        )

        return (
            point == left_limit
            or point == right_limit
        )

    def set_reach(self, reach_id):
        self._current_reach_id = reach_id
        self._current_profile_id = 0
        self.draw()

    def set_profile(self, profile_id):
        self._current_profile_id = profile_id
        self.update_current()

    def set_timestamp(self, timestamp):
        self._current_timestamp = timestamp
        self.update()

    def update(self):
        if not self._init:
            self.draw()

        self.update_bottom_with_bedload()
        self.update_water_elevation()

        self.update_idle()

    def update_water_elevation(self):
        reach = self.results.river.reach(self._current_reach_id)
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()

        water_z = list(
            map(
                lambda p: p.get_ts_key(
                    self._current_timestamp, "Z"
                ),
                reach.profiles
            )
        )

        self.water[0].set_data(
            kp, water_z
        )

        self.water_fill.remove()
        self.water_fill = self.canvas.axes.fill_between(
            kp, self._river_bottom, water_z,
            color=self.color_plot_river_water_zone,
            alpha=0.7, interpolate=True
        )

    def update_current(self):
        reach = self.results.river.reach(self._current_reach_id)
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()
        z_max = reach.geometry.get_z_max()
        cid = self._current_profile_id

        self.profile.set_data(
            [kp[cid], kp[cid]],
            [z_max[cid], z_min[cid]]
        )
        self.canvas.figure.canvas.draw_idle()

    def update_bottom_with_bedload(self):
        reach = self.results.river.reach(self._current_reach_id)
        kp = reach.geometry.get_kp()
        z_min = reach.geometry.get_z_min()

        initial_sl = self._initial_sl
        current_sl = self.sl_compute_current(reach)

        max_sl_num = reduce(
            lambda acc, sl: max(acc, len(sl)),
            current_sl, 0
        )

        sl_init, sl = self.sl_completed_layers(
            initial_sl, current_sl, max_sl_num
        )

        z_sl = self.sl_apply_z_min_to_initial(sl_init, z_min)
        d_sl = self.sl_compute_diff(sl_init, sl)

        final_z_sl = self.sl_apply_diff(reach, z_sl, d_sl)
        final_z_sl = list(reversed(final_z_sl))

        for i, z in enumerate(final_z_sl):
            self.line_kp_sl[i].set_data(kp, z)

        self._river_bottom = final_z_sl[-1]