app.py

"""Copyright (c) 2024 VIKTOR B.V.
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit
persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
Software.
VIKTOR B.V. PROVIDES THIS SOFTWARE ON AN "AS IS" BASIS, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

import json
from typing import Optional

import viktor as vkt

from gis_functions import get_download
from gis_functions import get_gdf
from gis_functions import set_filter_attributes
from parametrization import Parametrization


class Controller(vkt.ViktorController):
    label = "GIS-app"
    parametrization = Parametrization(width=30)

    @vkt.GeoJSONAndDataView("Map", duration_guess=1)
    def get_geojson_view(self, params, **kwargs) -> vkt.GeoJSONAndDataResult:
        """Show all the map elements and data results"""
        gdf = get_gdf(params.shape_input.shapefile_upload, params.shape_input.data_source, params.styling)
        geojson = json.loads(gdf.to_json())

        # Add labels to the map
        if params.shape_input.data_source == "Sample data":
            gdf_labels = gdf.dropna().copy()
            gdf_labels["label_geometry"] = gdf_labels.representative_point()
            labels = [
                vkt.MapLabel(point.y, point.x, str(gdf_labels.labels.iloc[index]), 3)
                for index, point in enumerate(gdf_labels.label_geometry)
            ]
        else:  # Create an empty placeholder for the labels
            gdf_labels = gdf.copy()
            gdf_labels["label_geometry"] = gdf_labels.representative_point()
            labels = [vkt.MapLabel(gdf_labels.label_geometry[0].x, gdf_labels.label_geometry[0].x, " ", 20)]

        # Set filter
        if params.attributes.set_filter:
            gdf_filtered = set_filter_attributes(gdf, params.attributes)
            geojson = json.loads(gdf_filtered.to_json())

        # Get and compare results for selected features
        if params.attribute_results:
            selected_features = params.attribute_results
            try:
                gdf_selected = gdf.loc[selected_features]
            except KeyError:
                raise vkt.UserError(
                    "Selection from sample data is still in memory. Please restart the app to clear " "the database."
                )
            if params.compare.field_name == params.compare.selected_value:
                raise vkt.UserError("Field names and compare for values cannot be the same. Please change this value.")
            gdf_combined = gdf_selected[[params.compare.field_name, params.compare.selected_value]].reset_index()
            gdf_combined = gdf_combined.sort_values(by=[params.compare.selected_value], ascending=False)
            attribute_results = vkt.DataGroup(
                vkt.DataItem(f"**{params.compare.field_name}**", f"**{params.compare.selected_value}**"),
                *[
                    vkt.DataItem(str(field_name), str(value))
                    for field_name, value in zip(
                        gdf_combined[params.compare.field_name].tolist(),
                        gdf_combined[params.compare.selected_value].tolist(),
                    )
                ],
            )
        else:  # Create an empty placeholder for results
            attribute_results = vkt.DataGroup(vkt.DataItem("Compare by ranking: Please select shapes to compare", ""))

        return vkt.GeoJSONAndDataResult(geojson, attribute_results, labels)

    def compare_attributes(self, event: Optional[vkt.InteractionEvent], **kwargs) -> vkt.SetParamsResult:
        """Returns the index of the selected features"""
        if event:
            selected_features = [int(value) for value in event.value]
            return vkt.SetParamsResult({"attribute_results": selected_features})

    def download_geopackage(self, params, **kwargs) -> vkt.DownloadResult:
        """Download selected results to a geopackage"""
        gdf = get_gdf(params.shape_input.shapefile_upload, params.shape_input.data_source, params.styling)
        if params.attributes.set_filter:
            gdf = set_filter_attributes(gdf, params.attributes)

        gdf.drop(columns=["description", "fill"], inplace=True)
        if gdf.crs.to_epsg() != params.download.output_crs:  # Check if output CRS is different from input CRS
            if params.download.output_crs == "Other":  # Use any EPSG coordinate system
                gdf = gdf.to_crs(params.download.output_crs_other)
            else:  # Use one of the predefined coordinate systems
                gdf = gdf.to_crs(params.download.output_crs)
        download_file, file_name = get_download(params.download.output_format_options, gdf)

        return vkt.DownloadResult(download_file, file_name)