g_s_various_functions.py

# -*- coding: utf-8 -*-
"""
/***************************************************************************
 GenerateSwmmInp
                                 A QGIS plugin
 This plugin generates SWMM Input files
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
                              -------------------
        begin                : 2021-07-09
        copyright            : (C) 2021 by Jannik Schilling
        email                : jannik.schilling@posteo.de
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   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 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""

__author__ = 'Jannik Schilling'
__date__ = '2023-05-09'
__copyright__ = '(C) 2021 by Jannik Schilling'

import pandas as pd
from qgis.PyQt.QtCore import QTime, QDate
from qgis.core import (
    QgsWkbTypes,
    QgsProcessingException
)
from .g_s_defaults import (
    def_tables_dict,
    annotation_field_name
)

# Export
# geometry functions
def check_nan(replace_lst):
    pass


def use_z_if_available(
    df,
    coords,
    use_z_bool,
    feedback,
    geom_type='Points',
    layer_name=None
):
    """
    replaces Elevation or InOffset/OutOffset by Z_Coords+
    :param pd.DataFrame df
    :param pd.DataFrame / dict coords
    :param bool use_z_bool
    :param QgsProcessingFeedback feedback
    """
    if geom_type == 'lines':
        if use_z_bool:
            # if not na
            df['InOffset'] = [coords[l_name]['Z_Coord'].tolist()[0] for l_name in df['Name']]
            df['OutOffset'] = [coords[l_name]['Z_Coord'].tolist()[-1] for l_name in df['Name']]
        coords = {
            l_name: df_coord[
                ['X_Coord', 'Y_Coord']
            ] for l_name, df_coord in coords.items()
        }  # remove z
    else:
        if use_z_bool:
            # if not na
            df['Elevation'] = coords['Z_Coord']
        coords.drop("Z_Coord", axis=1, inplace=True)
    return df, coords

def get_coords_from_geometry(df):
    """
    extracts coords from any gpd.geodataframe
    :param pd.DataFrame df
    """
    geom_point_types = {
        'Point': 'simple',
        'PointM': 'simple',
        'PointZ': 'simple',
        'PointZM': 'simple'
    }
    geom_line_types = {
        'LineString': 'simple',
        'LineStringZ': 'simple',
        'LineStringZM': 'simple',
        'LineStringM': 'simple',
        'MultiLineString': 'multi',
        'MultiLineStringM': 'multi',
        'MultiLineStringZ': 'multi',
        'MultiLineStringZM': 'multi'
    }
    geom_polygon_types = {
        'Polygon': 'simple',
        'PolygonZ': 'simple',
        'PolygonM': 'simple',
        'PolygonZM': 'simple',
        'MultiPolygon': 'multi',
        'MultiPolygonM': 'multi',
        'MultiPolygonZ': 'multi',
        'MultiPolygonZM': 'multi'
    }
    point_t_names = list(geom_point_types.keys())
    line_t_names = list(geom_line_types.keys())
    polygon_t_names = list(geom_polygon_types.keys())

    # case: points
    if all(
        QgsWkbTypes.displayString(
            g_type.wkbType()
        ) in point_t_names for g_type in df.geometry
    ):
        extr_coords = [
            extract_xyz_from_simple_point(
                p_name,
                point_simple
            ) for p_name, point_simple in zip(
                df['Name'],
                df['geometry']
            )
        ]
        extr_coords_df = pd.DataFrame(
            extr_coords,
            columns=(
                ['Name', 'X_Coord', 'Y_Coord', 'Z_Coord']
            )
        )
        return extr_coords_df

    # case lines
    elif all(
        QgsWkbTypes.displayString(
            g_type.wkbType()
        ) in line_t_names for g_type in df.geometry
    ):
        return {na: extract_xy_from_line(line_geom) for line_geom, na in zip(df.geometry, df.Name)}

    # case polygons
    elif all(
        QgsWkbTypes.displayString(
            g_type.wkbType()
        ) in polygon_t_names for g_type in df.geometry
    ):
        return {na: extract_xy_from_area(polyg_geom) for polyg_geom, na in zip(df.geometry, df.Name)}
    else:
        raise QgsProcessingException(
            'Geometry type of one or more features could not be handled'
        )

def extract_xyz_from_simple_point(p_name, point_simple):
    """
    extracts x and y coordinates from a LineString
    :param str p_name
    :param QgsGeometry point_simple
    :return: tuple
    """
    qgs_point = [p for p in point_simple.parts()][0]
    x_coord = str(qgs_point.x())
    y_coord = str(qgs_point.y())
    z_coord = qgs_point.z()
    return p_name, x_coord, y_coord, z_coord


def extract_xy_from_line(line_geom):
    """
    extraxts xy from LineString or MultiLineString
    :return: pd.DataFrame
    """
    vertices_list = [p for p in line_geom.vertices()]
    extr_coords = [
            extract_xyz_from_simple_point(
                'nan',
                point_simple
            ) for point_simple in 
                vertices_list
        ]
    extr_coords_df = pd.DataFrame(
        extr_coords,
        columns=(
            ['Name', 'X_Coord', 'Y_Coord', 'Z_Coord']
        )
    )
    extr_coords_df.drop('Name', axis=1, inplace=True)
    return extr_coords_df

def extract_xy_from_area(geom_row):
    """
    extraxts xy from polygon geometries
    :return: pd.DataFrame
    """
    xy_list = [[str(v.x()), str(v.y())] for v in geom_row.vertices()]
    xy_df = pd.DataFrame(xy_list, columns=['X_Coord', 'Y_Coord'])
    return xy_df

# functions for data in tables
def get_curves_from_table(curves_raw, name_col):
    """
    generates curve data for the input file from tables (curve_raw)
    :param pd.DataFrame curve_raw
    :param str name_col
    """
    curve_types = list(def_tables_dict['CURVES']['tables'].keys())
    curve_dict = dict()
    for curve_type in curve_types:
        if curve_type in curves_raw.keys():
            curve_df = curves_raw[curve_type]
            if len(curve_df.columns) > 3:
                curve_df = curve_df[curve_df.columns[:3]]
            curve_df = curve_df[curve_df[name_col] != ";"]
            curve_df = curve_df[pd.notna(curve_df[name_col])]
            if curve_df.empty:
                pass
            else:
                curve_df.set_index(keys=[name_col], inplace=True)
                for i in curve_df.index.unique():
                    curve = curve_df[curve_df.index == i]
                    curve = curve.reset_index(drop=True)
                    curve_dict[i] = {
                        'Name': i,
                        'Type': curve_type,
                        'frame': curve
                    }
    return (curve_dict)


def get_patterns_from_table(patterns_raw, name_col):
    """
    generates a pattern dict for the input file from tables (patterns_raw)
    :param pd.DataFrame patterns_raw
    :param str name_col
    """
    pattern_types = def_tables_dict['PATTERNS']['tables'].keys()
    pattern_dict = {}
    for pattern_type in pattern_types:
        pattern_cols = def_tables_dict['PATTERNS']['tables'][pattern_type].keys()
        pattern_df = patterns_raw[pattern_type]
        check_columns('Patterns Table', pattern_cols, pattern_df.columns)
        pattern_df = pattern_df[pattern_df[name_col] != ";"]
        pattern_df = pattern_df[pd.notna(pattern_df[name_col])]
        if pattern_df.empty:
            pass
        else:
            pattern_df.set_index(keys=[name_col], inplace=True)
            for i in pattern_df.index.unique():
                pattern = pattern_df[pattern_df.index == i]
                pattern = pattern.drop(columns=pattern.columns[0])
                pattern = pattern.reset_index(drop=True)
                pattern_dict[i] = {
                    'Name': i,
                    'Type': pattern_type,
                    'Factors': pattern
                }
    return (pattern_dict)


def adjust_datetime(
    dt_list,
    dt_type,
    str_output_format,
    ts_name,
    feedback
):
    """
    converts time values (tries different formats) into another time string
    :param list or series dt_list: column in which the date or time is written
    :param str dt_type: "Date" or "Time"
    :param str str_output_format
    :param str ts_name
    :param QgsProcessingFeedback feedback
    """
    dt_formats_dict = {
        'Date': ['yyyy-MM-dd', 'dd/MM/yyyy', 'dd.MM.yyyy'],
        'Time': ['HH:mm:ss', 'HH:mm', 'HH']
    }
    dt_formats = dt_formats_dict[dt_type]
    if all([type(dt_val) in [QDate, QTime] for dt_val in dt_list]):
        dt_val_list = [dt_val.toString(str_output_format) for dt_val in dt_list]
    else:
        dt_list = [str(dt_val) for dt_val in dt_list]
        if dt_type == 'Date':
            for d_f in dt_formats:
                dt_val_list = [QDate.fromString(dt_val, d_f) for dt_val in dt_list]
                if not any([x.isNull() for x in dt_val_list]):
                    break
        else:
            for d_f in dt_formats:
                dt_val_list = [QTime.fromString(dt_val, d_f) for dt_val in dt_list]
                if not any([x.isNull() for x in dt_val_list]):
                    break
        if not any([x.isNull() for x in dt_val_list]):
            dt_val_list = [dt_val.toString(str_output_format) for dt_val in dt_val_list]
            feedback.pushWarning(
                'Timeseries \"'+ts_name+'\" '+dt_type+'column was derived from strings (assumed format: '+d_f
            )
        else:
            raise QgsProcessingException(str(ts_name)+': column '+dt_type+' could not be converted properly. Tested formats: '+dt_formats)
    return dt_val_list


def get_timeseries_from_table(ts_raw, name_col, feedback):
    """
    generates a timeseries dict for the input file from tables (ts_raw)
    :param pd.DataFrame ts_raw
    :param str name_col
    :param QgsProcessingFeedback feedback
    """
    ts_dict = dict()
    ts_raw = ts_raw[ts_raw[name_col] != ";"]
    # warning for deprecated format:
    if ('Type' in ts_raw.columns) and ('Format' in ts_raw.columns):
        feedback.reportError(
            'Warning: The columns \"Type\" and \"Format\" '
            + 'are not used any longer in future versions of the plugin. '
            + 'Creating rain gages from timeseries only is deprecated. '
            + 'Please create a rain gage layer instead. You can get an '
            + 'examplary layer from the default data set or have a look '
            + 'at the documentation file.'
        )
    if ts_raw.empty:
        pass
    else:
        for ts_name in ts_raw[name_col].unique():
            ts_df = ts_raw[ts_raw[name_col] == ts_name]
            if 'File_Name' in ts_raw.columns and not all(pd.isna(ts_df['File_Name'])):  # external time series
                ts_df['Date'] = 'FILE'
                ts_df['Time'] = ts_df['File_Name']
                ts_df['Value'] = ''
            else:
                if sum(pd.isna(ts_df['Date'])) > 0:
                    # handes missing dates
                    if not all(pd.isna(ts_df['Date'])):
                        feedback.pushWarning(
                                'Warning: At least one date in the timeseries file is missing. Date will be set to start date')
                    ts_df['Date'] = ''
                else: 
                    ts_df['Date'] = adjust_datetime(
                        ts_df['Date'],
                        'Date',
                        'MM/dd/yyyy',
                        ts_name,
                        feedback = None
                    )
                ts_df['Time'] = adjust_datetime(
                    ts_df['Time'],
                    'Time',
                    'HH:mm',
                    ts_name,
                    feedback
                )
            if annotation_field_name in ts_df.columns:
                ts_annotation = ts_df[annotation_field_name].fillna('').unique()[0]
            else:
                ts_annotation = ''
            ts_dict[ts_name] = {
                'Name': ts_name,
                'TimeSeries': ts_df[['Name', 'Date', 'Time', 'Value']],
                'Annotations': ts_annotation
            }
    return (ts_dict)


# errors and feedback
def check_deprecated(
    swmm_data_file,
    swmm_section,
    df,
    cols_deprecated,
    feedback
):
    """
    :param str swmm_data_file
    :param str swmm_section
    :param pd.DataFrame df
    :param dic cols_deprecated: e.g. {'DeprecatedName': 'NewName'}
    :param QgsProcessingFeedback feedback
    """
    for dep_col in cols_deprecated.keys():
        if dep_col in df.columns:
            feedback.pushWarning(
                'Warning: usage of columns name \"' + dep_col +'\" in section '
                + swmm_section
                + ' is deprecated and will be removed in future versions of the plugin. Please use \"'
                + cols_deprecated[dep_col] + '\" instead.'
            )
            df = df.rename(columns={dep_col: cols_deprecated[dep_col]})
    return df
        
        
        
def check_columns(
    swmm_data_file,
    cols_expected,
    cols_in_df
):
    """
    checks if all columns are in a dataframe
    :param str swmm_data_file
    :param list cols_expected
    :param list cols_in_df
    """
    missing_cols = [x for x in cols_expected if x not in cols_in_df]
    if len(missing_cols) == 0:
        pass
    else:
        err_message = 'Missing columns in '+swmm_data_file+': '+', '.join(missing_cols)
        err_message = err_message+'. Please add columns or check if the correct file/layer was selected. '
        err_message = err_message+'For further advice regarding columns, read the documentation file in the plugin folder.'
        raise QgsProcessingException(err_message)