import json
import os
from datetime import timedelta, timezone
from enum import Enum
from pathlib import Path
from typing import Callable, Union

import numpy as np
import pandas
from qgis.core import (
    QgsApplication,
    QgsCoordinateReferenceSystem,
    QgsCoordinateTransform,
    QgsMapLayerProxyModel,
    QgsPointXY,
    QgsProject,
    QgsRasterLayer,
    QgsSettings,
)
from qgis.PyQt import QtCore, uic
from qgis.PyQt.QtGui import QColor, QFont, QIcon, QPixmap
from qgis.PyQt.QtWidgets import (
    QComboBox,
    QDialog,
    QFileDialog,
    QHBoxLayout,
    QInputDialog,
    QLabel,
    QLineEdit,
    QMessageBox,
    QPlainTextEdit,
    QPushButton,
    QStyledItemDelegate,
    QTableView,
    QTableWidget,
    QTableWidgetItem,
    QVBoxLayout,
    QWidget,
)

from openlog.__about__ import DIR_PLUGIN_ROOT
from openlog.core.assay_interpolation import (
    ExtendedAssayColumn,
    ExtendedAssayInterpolation,
)
from openlog.datamodel.assay.detection_limit import AssayColumnDetectionLimit
from openlog.datamodel.assay.generic_assay import AssaySeriesType, SphericalType
from openlog.datamodel.assay.spherical_assay import SphericalDefinition
from openlog.datamodel.assay.uncertainty import AssayColumnUncertainty, UncertaintyType
from openlog.datamodel.connection.openlog_connection import OpenLogConnection
from openlog.gui.import_assay.utils import sanitize_sql_name
from openlog.gui.utils.column_definition import ColumnDefinition
from openlog.gui.utils.column_definition_tablemodel import (
    SPECIAL_COLLAR_COLUMNS,
    SPECIAL_COLUMNS,
    SPECIAL_SURVEY_COLUMNS,
    ColumnDefinitionTableModel,
)
from openlog.gui.utils.mdl_assay_column_categorie import (
    AssayColumnCategoryTableModel,
    CategoryPolicy,
)
from openlog.gui.utils.mdl_assay_column_detection import AssayColumnDetectionLimitModel
from openlog.gui.utils.mdl_assay_column_uncertainty import (
    AssayColumnUncertaintyTableModel,
)
from openlog.gui.utils.mdl_assay_spherical_data import SphericalDataTableModel
from openlog.plugins.manager import get_plugin_manager
from openlog.toolbelt import encoding_utils
from openlog.toolbelt.log_handler import PlgLogger

CRS_KEY = "/crs"
DATE_FORMAT_KEY = "/date_format"
TIME_FORMAT_KEY = "/time_format"
COLUMN_LIST_KEY = "/column_list"
COLUMN_MAPPING_KEY = "/column_mapping"


class BigTextDialog(QDialog):
    def __init__(self, text, parent=None):
        super().__init__(parent)

        self.setWindowTitle("Input error")
        layout = QVBoxLayout(self)

        # widget efficace pour grand texte
        self.text_edit = QPlainTextEdit(self)
        self.text_edit.setPlainText(text)
        self.text_edit.setReadOnly(True)

        layout.addWidget(self.text_edit)

        btn_close = QPushButton("Close")
        btn_close.clicked.connect(self.accept)
        layout.addWidget(btn_close)


class UncertaintyTypeItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for column mapping definition

        Args:
            parent:
        """
        super().__init__(parent)
        self.available_options = [
            "None",
            "Interval",
            "Boundary pair",
            "Boundary quad",
        ]

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for column mapping definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available column for mapping

        """
        editor = QComboBox(parent)
        optional = True

        editor.addItems(self.available_options)
        editor.setCurrentIndex(0)
        widget = self.parent()
        if widget.already_created:
            editor.setEnabled(False)
        return editor


class ColumnMappingItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for column mapping definition

        Args:
            parent:
        """
        super().__init__(parent)
        self.available_columns = []  # Define columns available in file
        self.elevation_options = [
            "From DTM",
            "Set constant",
        ]  # additional options for elevation
        self.dip_azimuth_options = ["Set constant"]

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for column mapping definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available column for mapping

        """
        editor = QComboBox(parent)
        editor.setEditable(True)
        # settings for QCombobox adapting width to contents when opening list
        editor.setFixedWidth(option.rect.width())

        def adjust_popup():
            view = editor.view()
            width = view.sizeHintForColumn(0) + 30
            view.setMinimumWidth(width)

        editor.showPopup = lambda _orig=editor.showPopup: (adjust_popup(), _orig())[1]
        optional = True
        fixed = False
        name = ""
        col_def = []
        if isinstance(index.model(), ColumnDefinitionTableModel):
            optional = index.siblingAtRow(ColumnDefinitionTableModel.COLUMN_ROW).data(
                ColumnDefinitionTableModel.OPTIONAL_ROLE
            )
            fixed = index.siblingAtRow(ColumnDefinitionTableModel.COLUMN_ROW).data(
                ColumnDefinitionTableModel.FIXED_ROLE
            )
            name = index.siblingAtRow(ColumnDefinitionTableModel.COLUMN_ROW).data(
                ColumnDefinitionTableModel.COLUMN_ROW
            )
            col_def = index.model().get_column_definition()

        # behavior for uncertainties definition
        if (
            isinstance(index.model(), AssayColumnUncertaintyTableModel)
            and index.column() > AssayColumnUncertaintyTableModel.UNCERTAINTY_TYPE
        ):
            model = index.model()
            defined_type = model.data(
                model.index(
                    index.row(), AssayColumnUncertaintyTableModel.UNCERTAINTY_TYPE
                )
            )
            editor.setEnabled(False)
            editor.addItem("")

            if defined_type == "Interval":
                if index.column() == AssayColumnUncertaintyTableModel.WIDE_INTERVAL:
                    editor.setEnabled(True)
                    editor.addItems(self.available_columns)
            elif defined_type == "Boundary pair":
                if index.column() in (
                    AssayColumnUncertaintyTableModel.UPPER_OUTTER_BOUNDARY,
                    AssayColumnUncertaintyTableModel.LOWER_OUTTER_BOUNDARY,
                ):
                    editor.setEnabled(True)
                    editor.addItems(self.available_columns)
            elif defined_type == "Boundary quad":
                if index.column() in (
                    AssayColumnUncertaintyTableModel.UPPER_OUTTER_BOUNDARY,
                    AssayColumnUncertaintyTableModel.LOWER_OUTTER_BOUNDARY,
                    AssayColumnUncertaintyTableModel.UPPER_INNER_BOUNDARY,
                    AssayColumnUncertaintyTableModel.LOWER_INNER_BOUNDARY,
                ):
                    editor.setEnabled(True)
                    editor.addItems(self.available_columns)

            return editor

        # elevation column
        if (
            index.column() in [3, 7]
            and len(col_def) > 3
            and col_def[3].column == "Elevation*"
        ):

            for i, item in enumerate(self.elevation_options):
                editor.addItem(item)
                font = QFont("default", italic=True)
                editor.setItemData(i, font, QtCore.Qt.FontRole)

        # dip/azimuth column
        if (
            index.column() in [9, 10]
            and len(col_def) > 9
            and col_def[9].column == "Dip"
        ):

            for i, item in enumerate(self.dip_azimuth_options):
                editor.addItem(item)
                font = QFont("default", italic=True)
                editor.setItemData(i, font, QtCore.Qt.FontRole)

        # is spherical, disable column mapping
        if (
            len(col_def) > index.column()
            and col_def[index.column()].series_type == AssaySeriesType.SPHERICAL
        ):
            editor.setEnabled(False)

        if name not in SPECIAL_COLUMNS:
            editor.addItem("")
        self.available_columns = [str(col) for col in self.available_columns]
        editor.addItems(self.available_columns)
        return editor

    def paint(self, painter, option, index):
        # set item in red if not a column from csv
        value = index.data(QtCore.Qt.DisplayRole)
        is_invalid = value not in self.available_columns

        if is_invalid:
            option = option
            option.palette.setColor(option.palette.Text, QColor("red"))

        super().paint(painter, option, index)


class UpdatePolicyItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for database update options.

        Args:
            parent:
        """
        super().__init__(parent)
        self.widget = parent
        self.available_options = [
            "Overwrite",  # file wins
            "Ignore",  # don't update database
            "Merge from DB",  # fill database NULL with file values
            "Merge to DB",  # fill file NULL with database values + overwrite
        ]

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for column mapping definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available column for mapping

        """
        editor = QComboBox(parent)
        editor.addItems(self.available_options)

        return editor

    def setEditorData(self, editor, index):

        editor.setCurrentText("Ignore")
        editor.setEnabled(False)
        col = index.column()
        col_def = index.model().get_column_definition()
        name = col_def[col].column
        mapping = col_def[col].mapping
        series_type = col_def[col].series_type
        if self.widget.wizard_type == "assay":
            if name not in SPECIAL_COLUMNS:
                if mapping == "":
                    editor.setCurrentText("Ignore")
                    editor.setEnabled(False)
                    if series_type == AssaySeriesType.SPHERICAL:
                        editor.setCurrentText("Overwrite")
                        editor.setEnabled(True)
                else:
                    editor.setCurrentText("Overwrite")
                    editor.setEnabled(True)

        elif self.widget.wizard_type == "collar":
            if mapping == "":
                editor.setCurrentText("Ignore")
                editor.setEnabled(False)
                return
            if name not in SPECIAL_COLLAR_COLUMNS.keys():
                editor.setCurrentText("Overwrite")
                editor.setEnabled(True)
            else:
                editor.clear()
                editor.addItems(SPECIAL_COLLAR_COLUMNS.get(name))
                editor.setEnabled(True)

        elif self.widget.wizard_type == "survey":
            if mapping == "":
                editor.setCurrentText("Ignore")
                editor.setEnabled(False)
                return
            if name not in SPECIAL_SURVEY_COLUMNS.keys():
                editor.setCurrentText("Overwrite")
                editor.setEnabled(True)
            else:
                editor.clear()
                editor.addItems(SPECIAL_SURVEY_COLUMNS.get(name))
                editor.setEnabled(True)


class SphericalMappingItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for spherical assay mapping definition

        Args:
            parent:
        """
        super().__init__(parent)
        self.available_columns = []  # Define columns available in file

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for column mapping definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available column for mapping

        """
        editor = QComboBox(parent)
        editor.setEditable(True)

        editor.addItems(self.available_columns)
        return editor

    def paint(self, painter, option, index):
        # set item in red if not a column from csv
        value = index.data(QtCore.Qt.DisplayRole)
        is_invalid = value not in self.available_columns

        if is_invalid:
            option = option
            option.palette.setColor(option.palette.Text, QColor("red"))

        super().paint(painter, option, index)


class PolarityMappingItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for column mapping definition

        Args:
            parent:
        """
        super().__init__(parent)
        self.available_columns = []  # Define columns available in file
        self.other_options = ["Reverse", "Normal", "Unknown"]

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for column mapping definition + other options

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available column for mapping

        """
        editor = QComboBox(parent)

        for index, item in enumerate(self.other_options):
            editor.addItem(item)
            font = QFont("default", italic=True)
            editor.setItemData(index, font, QtCore.Qt.FontRole)

        editor.addItems(self.available_columns)
        return editor


class SeriesTypeItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for series type definition

        Args:
            parent:
        """
        super().__init__(parent)

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for series type definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available series type

        """
        editor = QComboBox(parent)

        editor.addItems(
            [e.name for e in AssaySeriesType if e != AssaySeriesType.PLUGIN]
        )
        return editor


class CategoriesValidationType(Enum):
    APPEND = "append"  # read categories is append to current categories data
    RESTRICT = "restrict"  # read categories must be in current categories data
    REMOVE = "remove"  # data with categories not in current categories data are removed


class CategoriesValidationTypeItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for categories validation type definition

        Args:
            parent:
        """
        super().__init__(parent)

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for series type definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available series type

        """
        editor = QComboBox(parent)

        editor.addItems([e.name for e in CategoriesValidationType])
        return editor


class SphericalTypeItemDelegate(QStyledItemDelegate):
    def __init__(self, parent) -> None:
        """
        QStyledItemDelegate for spherical type definition

        Args:
            parent:
        """
        super().__init__(parent)

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for spherical type definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available spherical type

        """
        editor = QComboBox(parent)

        editor.addItems([e.name for e in SphericalType])
        return editor


class CategoriesTableItemDelegate(QStyledItemDelegate):
    def __init__(self, parent, openlogconnection: OpenLogConnection = None) -> None:
        """
        QStyledItemDelegate for categorie definition

        Args:
            parent:
        """
        super().__init__(parent)
        self.openlogconnection = openlogconnection

    def createEditor(self, parent, option, index) -> QWidget:
        """
        Create a QComboBox for series type definition

        Args:
            parent: QWidget
            option: QStyleOptionViewItem
            index: QModelIndex

        Returns: QComboBox with available series type

        """
        editor = QComboBox(parent)
        editor.setEditable(True)
        if self.openlogconnection:
            editor.addItems(
                [
                    c.name
                    for c in self.openlogconnection.get_categories_iface().get_available_categories_table()
                ]
            )
        return editor


class DelimitedTextImportWidget(QWidget):
    red_flag_signal = QtCore.pyqtSignal()

    def __init__(
        self,
        parent=None,
        openlog_connection: OpenLogConnection = None,
        wizard_type: str = "assay",
    ) -> None:
        """
        Widget to import panda dataframe from delimited text.

        Wanted dataframe column must be defined with set_expected_column.

        Dataframe can be retrieved with get_dataframe, dataframe column name will match column
        name defined in set_expected_column.

        User can define column mapping inside widget.

        A widget for CRS selection can be enabled with enable_crs_selection (off by default)


        Args:
            parent:
            openlog_connection : OpenLogConnection instance
            wizard_type: could be collar, survey or assay
        """
        super().__init__(parent)
        self._openlog_connection = openlog_connection
        self.wizard_type = wizard_type
        # translation

        self.log = PlgLogger().log

        uic.loadUi(
            os.path.dirname(os.path.realpath(__file__))
            + "/delimited_text_import_widget.ui",
            self,
        )

        self.ags_reader = (
            get_plugin_manager().get_ags_reader_plugin().reader(wizard_type, parent)
        )
        self.ags_tables: dict = {}

        self.encoding_cbx.addItems(encoding_utils.ENCODING_LIST)
        self.encoding_cbx.setCurrentText("utf_8")
        self.encoding_cbx.currentIndexChanged.connect(self._source_updated)
        self.select_file_button.clicked.connect(self._select_file)
        self.delimiter_widget.delimiter_changed.connect(self._source_updated)

        # header
        self.header_row.valueChanged.connect(self._source_updated)

        # editable QCombobox
        self.project_layers = []
        if self._openlog_connection:
            self.project_layers = (
                self._openlog_connection.get_layers_iface().get_openlog_layer_names()
            )
        self.qgs_layers = [
            lyr.name()
            for lyr in QgsProject.instance().mapLayers().values()
            if hasattr(lyr, "fields")
        ]
        # remove project layers
        self.qgs_layers = [
            lyr for lyr in self.qgs_layers if lyr not in self.project_layers
        ]

        self.filename_edit.addItems([""] + self.qgs_layers)
        self.line_edit = QLineEdit(self)
        self.filename_edit.setLineEdit(self.line_edit)
        self.line_edit.editingFinished.connect(self._update_table_and_fields)
        self.filename_edit.currentIndexChanged.connect(self._source_updated)
        self.enable_crs_selection(False)
        self.enable_date_format_selection(False)
        self.dateformat_selection_widget.format_changed.connect(
            self._update_table_and_fields
        )
        # store z column mapping and value
        self._z_col_mapping = ""
        self.z_col_constant_value = None
        self._planned_z_col_mapping = ""
        self.planned_z_col_constant_value = None

        # store dip and azimuth mapping and value
        self._dip_col_mapping = ""
        self._dip_constant_value = None
        self._azimuth_col_mapping = ""
        self._azimuth_constant_value = None

        # dip and azimuth columns
        self._dip_col = ""
        self._azimuth_col = ""

        self._column_conversion = {}

        self._init_column_definition_section()
        self._init_spherical_section()
        self._init_uncertainties_section()
        self._init_detection_section()
        self._init_categories_section()

        # Quick fix for column type definition : must be updated when column type will be defined by user
        self.data_is_valid = True

        self.button_frame.setVisible(False)

        # DTM selection for elevation calculation
        self._x_col = ""
        self._y_col = ""
        self._z_col = ""
        self._planned_x_col = ""
        self._planned_y_col = ""
        self._planned_z_col = ""
        self.dtm_layer_combobox.setFilters(QgsMapLayerProxyModel.RasterLayer)
        self.dtm_layer_combobox.setAllowEmptyLayer(True, "None")
        self.enable_elevation_from_dtm(False)
        self.dtm_layer_combobox.layerChanged.connect(self._dtm_layer_changed)

        # store state
        self.red_flag = True
        # is assay already exist in database ?
        self.already_created = False
        # store raw dataframe
        self.df = None

        # for extended assays
        self.is_extended_assay = None
        self.removed_rows = 0
        self.UPDATE_STATUS_COL = "update_status"

        # store error messages
        self.messages = {}

    def get_import_policies(self) -> dict:
        """
        For each defined column, return import policy.
        """
        # Uncertainties and detection limits follow main column policy if mapped
        column_policies = self.column_mapping_model.get_import_policies()
        uncertainty_policies = {}
        uncertainties = self.get_assay_column_uncertainty()
        for key, uncertainty in uncertainties.items():
            main_column_policy = column_policies.get(key)
            if uncertainty.upper_whisker_column != "":
                uncertainty_policies[
                    uncertainty.upper_whisker_column
                ] = main_column_policy
            if uncertainty.lower_whisker_column != "":
                uncertainty_policies[
                    uncertainty.lower_whisker_column
                ] = main_column_policy
            if uncertainty.upper_box_column != "":
                uncertainty_policies[uncertainty.upper_box_column] = main_column_policy
            if uncertainty.lower_box_column != "":
                uncertainty_policies[uncertainty.lower_box_column] = main_column_policy

        detection_policies = {}
        detection_limits = self.get_assay_column_detection_limit()
        for key, detection in detection_limits.items():
            main_column_policy = column_policies.get(key)
            if detection.detection_min_col != "":
                detection_policies[detection.detection_min_col] = main_column_policy
            if detection.detection_max_col != "":
                detection_policies[detection.detection_max_col] = main_column_policy

        result = column_policies | uncertainty_policies | detection_policies

        return result

    def _init_column_definition_section(self) -> None:

        self.column_mapping_item_delegate = ColumnMappingItemDelegate(self)
        self.update_policy_item_delegate = UpdatePolicyItemDelegate(self)
        self.column_mapping_model = ColumnDefinitionTableModel(self)
        self.col_def_table_view.setModel(self.column_mapping_model)
        self.col_def_table_view.setItemDelegateForRow(
            self.column_mapping_model.MAPPING_ROW, self.column_mapping_item_delegate
        )
        self.col_def_table_view.setItemDelegateForRow(
            self.column_mapping_model.SERIES_TYPE_ROW, SeriesTypeItemDelegate(self)
        )
        self.col_def_table_view.setItemDelegateForRow(
            self.column_mapping_model.DB_UPDATE_POLICY, self.update_policy_item_delegate
        )
        self.col_def_table_view.hideRow(self.column_mapping_model.DB_UPDATE_POLICY)
        self.col_def_table_view.horizontalHeader().setVisible(False)

        self.column_mapping_model.dataChanged.connect(self.update_result_table)
        self.column_mapping_model.columnNameChanged.connect(self._update_column_name)

        self.add_column_button.clicked.connect(lambda: self.add_column("", ""))
        self.remove_column_button.clicked.connect(self._remove_column)
        self.remove_column_button.setEnabled(False)

        self.add_column_button.setIcon(QgsApplication.getThemeIcon("mActionAdd.svg"))
        self.remove_column_button.setIcon(
            QgsApplication.getThemeIcon("mActionRemove.svg")
        )

        self.col_def_table_view.selectionModel().selectionChanged.connect(
            self._update_button_status
        )

    def _init_detection_section(self) -> None:

        # DETECTION LIMIT GROUPBOX
        self.gpx_detection_limit.setVisible(False)
        self.mdl_assay_detection_limit = AssayColumnDetectionLimitModel(self)
        self.mdl_assay_detection_limit.dataChanged.connect(self.update_result_table)
        self.detection_table_view.setItemDelegateForColumn(
            self.mdl_assay_detection_limit.MIN_DETECTION,
            self.column_mapping_item_delegate,
        )
        self.detection_table_view.setItemDelegateForColumn(
            self.mdl_assay_detection_limit.MAX_DETECTION,
            self.column_mapping_item_delegate,
        )

        self.detection_table_view.setModel(self.mdl_assay_detection_limit)

    def _init_spherical_section(self) -> None:

        # SPHERICAL DATA BOX
        self.spherical_mapping_item_delegate = SphericalMappingItemDelegate(self)
        self.polarity_mapping_item_delegate = PolarityMappingItemDelegate(self)
        self.gpx_spherical_data.setVisible(False)
        self.mdl_spherical_data = SphericalDataTableModel(self)
        self.mdl_spherical_data.dataChanged.connect(self.update_result_table)
        self.spherical_table_view.setModel(self.mdl_spherical_data)
        self.spherical_table_view.setItemDelegateForColumn(
            self.mdl_spherical_data.TYPE_COL, SphericalTypeItemDelegate(self)
        )

        self.spherical_table_view.setItemDelegateForColumn(
            self.mdl_spherical_data.DIP_COL, self.spherical_mapping_item_delegate
        )
        self.spherical_table_view.setItemDelegateForColumn(
            self.mdl_spherical_data.AZIMUTH_COL, self.spherical_mapping_item_delegate
        )
        self.spherical_table_view.setItemDelegateForColumn(
            self.mdl_spherical_data.POLARITY_COL, self.polarity_mapping_item_delegate
        )

        self.btn_spherical_help.setIcon(
            QIcon(":/images/themes/default/mActionHelpContents.svg")
        )
        self.btn_spherical_help.clicked.connect(self._spherical_help_display)
        self.spherical_help_dialog = self._creation_spherical_help_dialog()

    def _init_categories_section(self) -> None:

        # CATEGORIES GROUP BOX
        self.gpx_categories.setVisible(False)

        self.mdl_assay_column_category = AssayColumnCategoryTableModel(self)
        self.mdl_assay_column_category.dataChanged.connect(self.update_result_table)

        self.category_item_delegate = CategoriesTableItemDelegate(
            self, self._openlog_connection
        )
        self.categories_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_category.CATEGORIE_COLUMN,
            self.category_item_delegate,
        )
        self.category_validation_item_delegate = CategoriesValidationTypeItemDelegate(
            self
        )
        self.categories_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_category.VALIDATION_COLUMN,
            self.category_validation_item_delegate,
        )
        self.categories_table_view.setModel(self.mdl_assay_column_category)

    def _init_uncertainties_section(self) -> None:

        # UNCERTAINTY GROUP BOX
        self.gpx_uncertainty.setVisible(False)
        self.mdl_assay_column_uncertainty = AssayColumnUncertaintyTableModel(self)
        self.uncertainty_type_item_delegate = UncertaintyTypeItemDelegate(self)
        self.mdl_assay_column_uncertainty.dataChanged.connect(self.update_result_table)
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.WIDE_INTERVAL,
            self.column_mapping_item_delegate,
        )
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.LOWER_OUTTER_BOUNDARY,
            self.column_mapping_item_delegate,
        )
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.UPPER_OUTTER_BOUNDARY,
            self.column_mapping_item_delegate,
        )
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.LOWER_INNER_BOUNDARY,
            self.column_mapping_item_delegate,
        )
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.UPPER_INNER_BOUNDARY,
            self.column_mapping_item_delegate,
        )
        self.uncertainty_table_view.setItemDelegateForColumn(
            self.mdl_assay_column_uncertainty.UNCERTAINTY_TYPE,
            self.uncertainty_type_item_delegate,
        )
        self.uncertainty_table_view.setModel(self.mdl_assay_column_uncertainty)

        self._update_uncertainty_table_model_columns()
        self.btn_uncertainty_help.setIcon(
            QIcon(":/images/themes/default/mActionHelpContents.svg")
        )
        self.btn_uncertainty_help.clicked.connect(self._uncertainty_help_display)
        self.uncertainty_help_dialog = self._creation_uncertainty_help_dialog()

    def restore_settings(self, base_key: str):
        """
        Restore settings from QgsSetting()

        Args:
            base_key: base key for QgsSettings

        """
        settings = QgsSettings()
        date_format = settings.value(base_key + DATE_FORMAT_KEY, "")
        if date_format:
            self.set_date_format(date_format)

        time_format = settings.value(base_key + TIME_FORMAT_KEY, "")
        if time_format:
            self.set_time_format(time_format)

        epsg_crs_id = settings.value(base_key + CRS_KEY, None)
        if epsg_crs_id:
            crs = QgsCoordinateReferenceSystem()
            if crs.createFromString(epsg_crs_id):
                self.mQgsProjectionSelectionWidget.setCrs(crs)
        column_list = settings.value(base_key + COLUMN_LIST_KEY)
        if column_list:
            for col in column_list:
                mapping = settings.value(base_key + COLUMN_MAPPING_KEY + f"/{col}", "")
                self.column_mapping_model.set_column_mapping(col, mapping)

        self.delimiter_widget.restore_settings(base_key)

    def save_setting(self, base_key: str):
        """
        Store settings in QgsSettings()

        Args:
            base_key: base key for QgsSettings

        """
        settings = QgsSettings()
        settings.setValue(
            base_key + DATE_FORMAT_KEY,
            self.dateformat_selection_widget.get_date_format(),
        )
        settings.setValue(
            base_key + TIME_FORMAT_KEY,
            self.dateformat_selection_widget.get_time_format(),
        )
        settings.setValue(
            base_key + CRS_KEY, self.mQgsProjectionSelectionWidget.crs().authid()
        )
        column_mapping = self.get_column_definition()
        columns_names = [x.column for x in column_mapping]
        settings.setValue(base_key + COLUMN_LIST_KEY, list(columns_names))
        for mapping in column_mapping:
            settings.setValue(
                base_key + COLUMN_MAPPING_KEY + f"/{mapping.column}", mapping.mapping
            )

        self.delimiter_widget.save_setting(base_key)

    def get_file_path(self) -> Path:
        """
        Get file path of current selected file

        Returns: (Path) current selected file

        """
        return Path(self.filename_edit.lineEdit().text())

    def set_column_conversion(
        self, column_conversion: {str: Callable[[str, pandas.DataFrame], pandas.Series]}
    ):
        """
        Define conversion callable for columns

        Args:
            column_conversion: map of column name associated with callable for conversion
        """
        self._column_conversion = column_conversion

    def set_column_definition(self, columns: [ColumnDefinition]) -> None:
        """
        Define dataframe expected column names

        Args:
            columns: dataframe column names
        """
        # Define column mapping
        self.column_mapping_model.dataChanged.disconnect(self.update_result_table)
        self.column_mapping_model.set_column_definition(columns)
        self.column_mapping_model.dataChanged.connect(self.update_result_table)

        # Check if datetime column is available
        self.enable_date_format_selection(
            any(x.series_type == AssaySeriesType.DATETIME for x in columns)
        )

        fixed_cols = [c.column for c in columns if c.fixed]

        # Define categories
        column_categories = []
        if self._openlog_connection:
            for col in columns:
                if col.category_name:
                    column_categories.append(
                        self._create_category_policy(col.column, col.category_name)
                    )
        self.mdl_assay_column_category.set_assay_column_categories(column_categories)
        self.mdl_assay_column_category.fixed_columns = fixed_cols
        self.gpx_categories.setVisible(len(column_categories) > 0)

    def update_optionnal_widgets(self, assay_definition) -> None:
        """
        Used when importing data into an existing assay.
        Update the folowing widgets :
            - uncertainties
            - detection limits
        Args:
            - assay_definition : an AssayDefinition instance
        """

        if not self.already_created:
            self.col_def_table_view.hideRow(self.column_mapping_model.DB_UPDATE_POLICY)
            self.add_column_button.setEnabled(True)
            return

        self.add_column_button.setEnabled(False)
        self.col_def_table_view.showRow(self.column_mapping_model.DB_UPDATE_POLICY)

        if self.wizard_type != "assay":
            # enable add button for collar metadatas
            if self.wizard_type == "collar":
                self.add_column_button.setEnabled(True)
            return
        # uncertainties section
        d = {}

        for column_name, assay_column in assay_definition.columns.items():

            if assay_column.series_type == AssaySeriesType.NUMERICAL:
                d[column_name] = assay_column.uncertainty

        # add rows
        self._update_uncertainty_table_model_columns()
        self.mdl_assay_column_uncertainty.dataChanged.disconnect(
            self.update_result_table
        )
        self.mdl_assay_column_uncertainty.set_assay_column_uncertainty(d)
        self.mdl_assay_column_uncertainty.reinitialize_all_mapping_columns()
        self.gpx_uncertainty.setVisible(len(d) > 0)
        self.mdl_assay_column_uncertainty.dataChanged.connect(self.update_result_table)

        # detection limits
        wanted_detection = False
        self.mdl_assay_detection_limit.dataChanged.disconnect(self.update_result_table)
        for column_name, assay_column in assay_definition.columns.items():
            if assay_column.series_type != AssaySeriesType.NUMERICAL:
                continue
            detection = assay_column.detection_limit
            self.mdl_assay_detection_limit.add_assay_column_detection_limit(
                column=column_name, detection_limit=AssayColumnDetectionLimit()
            )
            # set cell fixed
            if detection.detection_min_col == "":
                self.mdl_assay_detection_limit.set_cell_uneditable(
                    self.mdl_assay_detection_limit.rowCount() - 1,
                    self.mdl_assay_detection_limit.MIN_DETECTION,
                )
            else:
                wanted_detection = True
            if detection.detection_max_col == "":
                self.mdl_assay_detection_limit.set_cell_uneditable(
                    self.mdl_assay_detection_limit.rowCount() - 1,
                    self.mdl_assay_detection_limit.MAX_DETECTION,
                )
            else:
                wanted_detection = True

        self.gpx_detection_limit.setVisible(wanted_detection)
        self.mdl_assay_detection_limit.dataChanged.connect(self.update_result_table)

        # spherical
        self.mdl_spherical_data.dataChanged.disconnect(self.update_result_table)
        d = {}
        for column_name, assay_column in assay_definition.columns.items():
            if assay_column.series_type != AssaySeriesType.SPHERICAL:
                continue
            d[column_name] = SphericalDefinition()

        self.mdl_spherical_data.set_spherical_assay(d)
        self.gpx_spherical_data.setVisible(len(d) > 0)
        self.mdl_spherical_data.dataChanged.connect(self.update_result_table)

    def get_column_definition(self) -> [ColumnDefinition]:
        """
        Get dataframe expected column names
        """
        res = self.column_mapping_model.get_column_definition(False)

        # Add category table name
        assay_column_category = (
            self.mdl_assay_column_category.get_assay_column_categories()
        )
        for column_def in res:
            col = column_def.column
            if col in assay_column_category:
                column_def.category_name = assay_column_category[col]
        return res

    def get_assay_column_uncertainty(self) -> {str: AssayColumnUncertainty}:
        """
        Return dict of AssayColumnUncertainty

        Returns: {str: AssayColumnUncertainty}

        """
        result = self.mdl_assay_column_uncertainty.get_assay_column_uncertainty()

        # Update uncertainty column name to avoid column name duplication
        columns = self.column_mapping_model.get_column_definition(True)
        assay_columns_name = [col.column for col in columns]

        for column, uncertainty in result.items():
            if uncertainty.get_uncertainty_type() == UncertaintyType.ONE_COLUMN:
                uncertainty.upper_whisker_column = self._get_unique_column_name(
                    column, assay_columns_name, "wide"
                )
            if (
                uncertainty.get_uncertainty_type() == UncertaintyType.TWO_COLUMN
                or uncertainty.get_uncertainty_type() == UncertaintyType.FOUR_COLUMN
            ):
                uncertainty.upper_whisker_column = self._get_unique_column_name(
                    column, assay_columns_name, "outer_max"
                )
                uncertainty.lower_whisker_column = self._get_unique_column_name(
                    column, assay_columns_name, "outer_min"
                )

            if uncertainty.get_uncertainty_type() == UncertaintyType.FOUR_COLUMN:
                uncertainty.upper_box_column = self._get_unique_column_name(
                    column, assay_columns_name, "inner_max"
                )
                uncertainty.lower_box_column = self._get_unique_column_name(
                    column, assay_columns_name, "inner_min"
                )
        return result

    def get_assay_column_detection_limit(self) -> {str: AssayColumnDetectionLimit}:
        """
        Return dict of AssayColumnDetectionLimit

        Returns: {str: AssayColumnDetectionLimit}

        """
        result = self.mdl_assay_detection_limit.get_assay_column_detection_limit()

        # Update detection column name to avoid column name duplication
        columns = self.column_mapping_model.get_column_definition(True)
        assay_columns_name = [col.column for col in columns]

        for column, detection in result.items():

            if detection.detection_min_col != "":
                detection.detection_min_col = self._get_unique_column_name(
                    column, assay_columns_name, "detlim_min"
                )
            if detection.detection_max_col != "":
                detection.detection_max_col = self._get_unique_column_name(
                    column, assay_columns_name, "detlim_max"
                )
        return result

    @staticmethod
    def _get_unique_column_name(
        init_col: str, assay_lower_columns_name_: [str], suffix: str = ""
    ) -> str:
        index_ = 1
        col = f"{init_col}_{suffix}"
        while col in assay_lower_columns_name_:
            col = f"{init_col}_{index_}"
            index_ = index_ + 1
        return col

    def get_used_category(self) -> []:
        """
        Return dict of category by column

        Returns: {str: str}

        """
        return list(
            set(self.mdl_assay_column_category.get_assay_column_categories().values())
        )

    def set_delimiter(self, delimiter: str) -> None:
        """
        Define delimiter for delimiter widget

        Args:
            delimiter: delimiter
        """
        self.delimiter_widget.set_delimiter(delimiter)

    def add_column(self, column: str = "", mapping: str = "") -> None:
        """
        Add a new column

        Args:
            column: (str) column name
            mapping: (str) column mapping

        """
        self.column_mapping_model.dataChanged.disconnect(self.update_result_table)
        self.column_mapping_model.add_column_definition(
            ColumnDefinition(
                column=column,
                mapping=mapping,
                series_type=AssaySeriesType.NUMERICAL,
                optional=True,
            )
        )
        self.column_mapping_model.dataChanged.connect(self.update_result_table)

    def set_button_layout_visible(self, visible: bool) -> None:
        """
        Change button layout visibility

        Args:
            visible: (bool) True to set button layout visible, False to set button layout invisible
        """
        self.button_frame.setVisible(visible)

    def _update_button_status(self) -> None:
        """
        Update button status from current selection. Used to define column remove enable

        """
        enable_delete = False
        for index in self.col_def_table_view.selectionModel().selectedIndexes():
            fixed = self.column_mapping_model.data(
                self.column_mapping_model.index(
                    self.column_mapping_model.COLUMN_ROW, index.column()
                ),
                QtCore.Qt.UserRole,
            )
            if not fixed:
                enable_delete = True

        self.remove_column_button.setEnabled(enable_delete)

    def _remove_column(self) -> None:
        """
        Remove column from current selection. Only not fixed column are deleted

        """
        cols = []
        for index in self.col_def_table_view.selectionModel().selectedIndexes():
            fixed = self.column_mapping_model.data(
                self.column_mapping_model.index(
                    self.column_mapping_model.COLUMN_ROW, index.column()
                ),
                QtCore.Qt.UserRole,
            )
            col = self.column_mapping_model.data(
                self.column_mapping_model.index(
                    self.column_mapping_model.COLUMN_ROW, index.column()
                )
            )
            if not fixed:
                cols.append(col)
        cols = set(cols)
        for col in cols:
            self.column_mapping_model.remove_column(col)
            self.mdl_assay_column_uncertainty.remove_column(col)
            self.mdl_assay_column_category.remove_column(col)
            self.update_result_table()

    def get_used_categories_values(self) -> {str: [str]}:
        """
        Get values imported of each category

        Returns: {str: [str]} list of values for each used category

        """
        result = {}
        df = self.get_dataframe()
        if df is not None:
            categories_map = (
                self.mdl_assay_column_category.get_assay_column_categories()
            )
            for col, category in categories_map.items():
                values = (
                    df[col].replace(to_replace="None", value=np.nan).dropna().to_list()
                )
                if category not in result:
                    result[category] = values
                else:
                    result[category] = result[category] + values
        return result

    def get_dataframe(
        self, nb_row: int = None, spherical: bool = False
    ) -> pandas.DataFrame:
        """
        Get panda dataframe from user selected file and column mapping

        Args:
            nb_row: read only a specific number of row (default None : all rows are read)
            spherical: if True, returns spherical columns in JSON-like format for import in database.

        Returns:
            panda dataframe with expected column

        """

        if self.df is not None:
            columns = self.column_mapping_model.get_column_definition(True)

            # remove Elevation from columns (treated after)
            columns = [
                col
                for col in columns
                if col.column not in [self._z_col, self._planned_z_col]
            ]

            # remove dip and azimuth from columns (treated after)
            columns = [
                col
                for col in columns
                if col.column not in [self._dip_col, self._azimuth_col]
            ]

            # Define uncertainty mapping from model (columns from file header)
            uncertainty_columns_mapping = []
            for (
                col,
                uncertainty,
            ) in (
                self.mdl_assay_column_uncertainty.get_assay_column_uncertainty().items()
            ):
                uncertainty_columns_mapping += uncertainty.get_uncertainty_columns()

            # Define column mapping
            assay_columns_mapping = [
                col.mapping for col in columns
            ] + uncertainty_columns_mapping

            # Get uncertainty columns from class method to get updated names from assay column
            uncertainty_columns = []
            for col, uncertainty in self.get_assay_column_uncertainty().items():
                uncertainty_columns += uncertainty.get_uncertainty_columns()

            assay_columns_name = [col.column for col in columns] + uncertainty_columns

            # Define detection mapping from model (columns from file header)
            detection_columns_mapping = []
            for (
                col,
                detection,
            ) in (
                self.mdl_assay_detection_limit.get_assay_column_detection_limit().items()
            ):
                detection_columns_mapping += detection.get_detection_columns()

            # Define column mapping
            assay_columns_mapping += detection_columns_mapping

            # Get detection columns from class method to get updated names from assay column
            detection_columns = []
            for col, detection in self.get_assay_column_detection_limit().items():
                detection_columns += detection.get_detection_columns()

            assay_columns_name += detection_columns

            # Get complete dataframe from file
            df = self._get_sample_dataframe(nb_row)

            # a mapping column doesn't exist in source dataframe
            if len([c for c in assay_columns_mapping if c not in df.columns]) > 0:
                self.red_flag = True
                self.red_flag_signal.emit()

            # remove final columns if source column don't exist
            assay_columns_name = [
                preview_col
                for preview_col, mapped_column in zip(
                    assay_columns_name, assay_columns_mapping
                )
                if mapped_column in df.columns
            ]

            # Get selected columns (only existing columns)
            df = df[[c for c in assay_columns_mapping if c in df.columns]]

            # Rename columns with wanted name
            df.columns = assay_columns_name

            # apply null values according no data
            for csv_col, preview_col in zip(assay_columns_mapping, assay_columns_name):
                pattern = [
                    column.no_data_pattern
                    for column in columns
                    if column.mapping == csv_col and column.column == preview_col
                ]
                if len(pattern) == 1:
                    df[preview_col] = df[preview_col].replace(pattern[0], None)

            # Check that all columns are defined and date conversion
            df = self._add_wanted_columns(df)

            # Convert data with current series type
            df = self._series_type_conversion(columns, df)

            # Convert data with column conversion
            df = self._user_columns_conversion(df)

            # Update value for elevation
            df = self._update_elevation(df=df, planned=False)
            df = self._update_elevation(df=df, planned=True)

            # update dip and azimuth values
            df = self._update_dip_azimuth(df=df)

            # return true spherical data
            if spherical:

                df = self._create_spherical_data(df)

            if self.is_extended_assay:
                value_cols = [
                    col
                    for col in df.columns
                    if col
                    not in [
                        self.parent()._HOLE_ID,
                        self.parent()._X_COL,
                        self.parent()._X_END_COL,
                    ]
                ]
                interpolation = ExtendedAssayInterpolation(
                    gap_resolution=self.parent().get_gap_data_resolution(),
                    overlap_resolution=self.parent().get_overlap_data_resolution(),
                    columns=ExtendedAssayColumn(
                        hole_id_col=self.parent()._HOLE_ID,
                        from_col=self.parent()._X_COL,
                        to_col=self.parent()._X_END_COL,
                        val_cols=value_cols,
                        update_status_col=self.UPDATE_STATUS_COL,
                    ),
                )
                try:
                    df = interpolation.interpolated_dataframe(df)
                    self.data_is_valid = True
                except ExtendedAssayInterpolation.InvalidColumnData:
                    self.data_is_valid = False

                # delete rows with 0 thickness
                nrow = df.shape[0]
                df = df[df[self.parent()._X_COL] != df[self.parent()._X_END_COL]]
                self.removed_rows = nrow - df.shape[0]
            return df
        else:
            return None

    def _create_spherical_data(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """
        Create real spherical columns to import into database.
        A spherical column is in a JSON-like string format : {dip: 0, azimuth: 0, polarity:0, type:'line'}
        """
        raw_df = self._get_sample_dataframe()
        sph_definitions = self.mdl_spherical_data.get_spherical_assay()

        # add fake columns for polarity constants
        for value, opt in enumerate(self.polarity_mapping_item_delegate.other_options):
            raw_df[opt] = value

        columns = self.get_column_definition()
        columns = [
            col for col in columns if col.series_type == AssaySeriesType.SPHERICAL
        ]
        for col in columns:
            col_name = col.column
            sph_def = sph_definitions[col_name]
            try:
                if sph_def.type_ == "LINE":

                    serie = raw_df.agg(
                        lambda x: {
                            "dip": float(x[sph_def.dip_col]),
                            "azimuth": float(x[sph_def.azimuth_col]),
                            "polarity": 0,
                            "type": "LINE",
                        },
                        axis=1,
                    )
                else:
                    serie = raw_df.agg(
                        lambda x: {
                            "dip": float(x[sph_def.dip_col]),
                            "azimuth": float(x[sph_def.azimuth_col]),
                            "polarity": float(x[sph_def.polarity_col]),
                            "type": "PLANE",
                        },
                        axis=1,
                    )

                serie = [json.dumps(elt) for elt in serie]
                # replace status message by JSON in df
                df[col_name] = serie
            except:
                df = df.drop(columns=[col_name])

        return df

    def _series_type_conversion(
        self, columns: [ColumnDefinition], df: pandas.DataFrame
    ) -> pandas.DataFrame:
        """
        Dataframe type conversion with defined columns series type

        Args:
            columns: [ColumnDefinition] column definition list
            df: input dataframe

        Returns: dataframe with column conversion

        """
        self.data_is_valid = True

        def _set_column_color_and_tooltip(col_: int, tooltip: str, color: str) -> None:
            self.column_mapping_model.dataChanged.disconnect(self.update_result_table)
            self.column_mapping_model.set_column_color_and_tooltip(
                col_, tooltip, QColor(color)
            )
            self.column_mapping_model.dataChanged.connect(self.update_result_table)

        remove_index = []
        for col_def in columns:
            col = col_def.column
            type_ = col_def.series_type.python_type()
            try:
                if col_def.series_type == AssaySeriesType.DATETIME:

                    tz = timezone(
                        timedelta(
                            hours=self.dateformat_selection_widget.get_time_offset()
                        )
                    )

                    defined_format = (
                        self.dateformat_selection_widget.get_datetime_format()
                    )
                    _format = None if defined_format == "auto" else defined_format

                    df[col] = (
                        pandas.to_datetime(
                            df[col],
                            format=_format,
                            errors="raise",
                        )
                        .dt.tz_localize(tz)
                        .dt.tz_convert(None)
                    )
                    df[col] = df[col].replace({pandas.NaT: None})
                elif col_def.series_type == AssaySeriesType.NOMINAL:
                    # dont convert None to str
                    df[col] = df[col].astype(type_)
                    df[col].replace("None", None, inplace=True)
                elif col_def.series_type == AssaySeriesType.NUMERICAL:
                    df[col] = df[col].replace({"": None})
                    # use pandas because castable values are casted and non-castable values are nan
                    df[col] = pandas.to_numeric(df[col], errors="coerce")
                elif col_def.series_type == AssaySeriesType.CATEGORICAL:
                    # dont convert None to str
                    df[col] = df[col].astype(type_)
                    df[col].replace("None", None, inplace=True)
                    #  Check category values
                    categories = (
                        self.mdl_assay_column_category.get_assay_column_categories()
                    )
                    categories_table = categories[col_def.column]
                    validation = self.mdl_assay_column_category.get_assay_column_category_validation(
                        col_def.column
                    )
                    current_cat = [
                        c.name
                        for c in self._openlog_connection.get_categories_iface().get_available_categories_table()
                    ]
                    if categories_table not in current_cat:
                        values = []
                    else:
                        cat = self._openlog_connection.get_categories_iface().get_categories_table(
                            categories_table
                        )
                        values = self._openlog_connection.get_categories_iface().get_available_categories(
                            cat
                        )
                    # For now if no values available allow all values
                    if not len(values):
                        validation = CategoriesValidationType.APPEND.name
                    if validation == CategoriesValidationType.RESTRICT.name:
                        df[col] = df[col].apply(lambda x: x if x in values else None)
                    elif validation == CategoriesValidationType.REMOVE.name:
                        df[col] = df[col].apply(lambda x: x if x in values else np.nan)
                        remove_index += df[df[col].isna()].index.to_list()
                elif col_def.series_type == AssaySeriesType.IMAGERY:
                    df[col] = df[col].astype(str)

                    def _convert_path(
                        imported_file_path: Path, image_path_str: str
                    ) -> Union[Path, None]:
                        # Check if file is available
                        image_path = Path(image_path_str)
                        if not image_path.exists() and imported_file_path.exists():
                            image_path = imported_file_path.parent / image_path_str
                        if image_path.exists():
                            return str(image_path)
                        else:
                            return None

                    df[col] = df[col].apply(
                        lambda x: _convert_path(self.get_file_path(), x)
                    )

                # Remove rows with mandatory values not defined
                if col in SPECIAL_COLUMNS:
                    remove_index += df[df[col].isna()].index.to_list()

                _set_column_color_and_tooltip(col, "", "black")
            except (ValueError, pandas.errors.ParserError) as exc:
                error = f"Invalid data in {col} column. {type_} expected. {exc}"
                _set_column_color_and_tooltip(col, error, "red")
                df[col] = np.nan
                self.data_is_valid = False

        df = df.drop(remove_index)
        return df

    def _update_dip_azimuth(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """
        Update dip and azimuth columns.
        """
        column_def = self.column_mapping_model.get_column_definition(False)

        dip_mapping = [
            col_def.mapping for col_def in column_def if col_def.column == self._dip_col
        ]
        azimuth_mapping = [
            col_def.mapping
            for col_def in column_def
            if col_def.column == self._azimuth_col
        ]

        if len(dip_mapping) == 1:
            dip_mapping = dip_mapping[0]
            if dip_mapping == "Set constant":
                if self._dip_col_mapping != "Set constant":
                    dip_value, done = QInputDialog.getDouble(
                        self, "Input Dialog", "Enter dip value:", min=-90, max=90
                    )

                    if done:
                        df[self._dip_col] = dip_value
                        self._dip_constant_value = dip_value

                else:
                    df[self._dip_col] = self._dip_constant_value

            elif dip_mapping != "":
                raw_df = self._get_sample_dataframe()
                # replace empty by 0
                raw_df[dip_mapping] = raw_df[dip_mapping].replace(
                    to_replace="", value="0"
                )
                try:
                    df[self._dip_col] = raw_df[dip_mapping].astype(float)
                except ValueError as e:
                    df[self._dip_col] = np.nan

            self._dip_col_mapping = dip_mapping

        if len(azimuth_mapping) == 1:
            azimuth_mapping = azimuth_mapping[0]
            if azimuth_mapping == "Set constant":
                if self._azimuth_col_mapping != "Set constant":
                    azimuth_value, done = QInputDialog.getDouble(
                        self, "Input Dialog", "Enter azimuth value:", min=0, max=360
                    )

                    if done:
                        df[self._azimuth_col] = azimuth_value
                        self._azimuth_constant_value = azimuth_value

                else:
                    df[self._azimuth_col] = self._azimuth_constant_value

            elif azimuth_mapping != "":
                raw_df = self._get_sample_dataframe()
                # replace empty by 0
                raw_df[azimuth_mapping] = raw_df[azimuth_mapping].replace(
                    to_replace="", value="0"
                )
                try:
                    df[self._azimuth_col] = raw_df[azimuth_mapping].astype(float)
                except ValueError as e:
                    df[self._azimuth_col] = np.nan

            self._azimuth_col_mapping = azimuth_mapping

        return df

    def _update_elevation(
        self, df: pandas.DataFrame, planned: bool
    ) -> pandas.DataFrame:
        """
        Update elevation column from selected DTM

        Args:
            df: input dataframe


        Returns: dataframe with calculated elevation column

        """
        x_column = self._planned_x_col if planned else self._x_col
        y_column = self._planned_y_col if planned else self._y_col
        z_column = self._planned_z_col if planned else self._z_col
        z_col_mapping = self._planned_z_col_mapping if planned else self._z_col_mapping
        z_col_constant_value = (
            self.planned_z_col_constant_value if planned else self.z_col_constant_value
        )

        # get mapping column
        column_def = self.column_mapping_model.get_column_definition(False)
        z_mapping = [
            col_def.mapping for col_def in column_def if col_def.column == z_column
        ]
        if len(z_mapping) == 0:
            return df

        z_mapping = z_mapping[0]

        df[z_column] = None
        if z_mapping == "From DTM":

            # Use isHidden instead of isVisible
            # Because isVisible check for current screen visibility
            dtm_layer = (
                self.dtm_layer_combobox.currentLayer()
                if not self.dtm_layer_combobox.isHidden()
                else QgsRasterLayer()
            )
            if (
                not df.empty
                and x_column
                and y_column
                and z_column
                and dtm_layer
                and dtm_layer.isValid()
            ):
                df[z_column] = df.apply(
                    lambda row: self._define_elevation(row, planned), axis=1
                )

        elif z_mapping == "Set constant":
            # avoid pop up dialog each time
            if z_col_mapping != "Set constant":
                z_value, done = QInputDialog.getDouble(
                    self, "Input Dialog", "Enter Z value:"
                )

                if done:
                    df[z_column] = z_value
                    if planned:
                        self.planned_z_col_constant_value = z_value
                    else:
                        self.z_col_constant_value = z_value

            else:
                df[z_column] = z_col_constant_value

        elif z_mapping != "":
            raw_df = self._get_sample_dataframe()
            # replace empty by 0
            raw_df[z_mapping] = raw_df[z_mapping].replace(to_replace="", value="0")
            try:
                df[z_column] = raw_df[z_mapping].astype(float)
            except ValueError as e:
                df[z_column] = np.nan

        if planned:
            self._planned_z_col_mapping = z_mapping
        else:
            self._z_col_mapping = z_mapping

        return df

    def _define_elevation(self, row, planned: bool):
        """
        Define elevation for current dataframe row from DTM layer and X/Y coordinate or available Z value

        Args:
            row: dataframe row

        Returns: Z value for dataframe row

        """
        x_column = self._planned_x_col if planned else self._x_col
        y_column = self._planned_y_col if planned else self._y_col
        z_column = self._planned_z_col if planned else self._z_col

        if (
            row[z_column] is None
            and row[x_column] is not None
            and row[y_column] is not None
        ):
            return self._get_z_value_from_dtm(QgsPointXY(row[x_column], row[y_column]))
        else:
            return row[z_column]

    def _get_z_value_from_dtm(self, point: QgsPointXY) -> float:
        """
        Calculate Z value from selected DTM

        Args:
            point:  QgsPointXY input point

        Returns: z value from DTM if available, None otherwise

        """
        tr = QgsCoordinateTransform(
            self.mQgsProjectionSelectionWidget.crs(),
            self.dtm_layer_combobox.currentLayer().crs(),
            QgsProject.instance(),
        )
        point_dtm = tr.transform(point)
        z_val, res = (
            self.dtm_layer_combobox.currentLayer().dataProvider().sample(point_dtm, 1)
        )
        if res:
            z = z_val
        else:
            self.log(
                "Can't define DTM value for point : {0}/{1}. z value used is None.".format(
                    point.x(), point.y()
                ),
                push=True,
                parent_location=self,
            )
            z = None
        return z

    def _create_spherical_preview(self, col: str) -> str:
        """
        Create status for spherical data preview.
        Args:
            col: column name

        Returns : Not defined or Valid or Values outside range or Wrong data type
        """

        # get spherical definitions
        sph_def = self.mdl_spherical_data.get_spherical_assay()[col]
        raw_df = self._get_sample_dataframe()
        # check if mapped column exists in source data, else set ""
        if sph_def.dip_col != "" and sph_def.dip_col not in raw_df.columns:
            sph_def.dip_col = ""
        if sph_def.azimuth_col != "" and sph_def.azimuth_col not in raw_df.columns:
            sph_def.azimuth_col = ""
        if sph_def.polarity_col != "" and sph_def.polarity_col not in raw_df.columns:
            sph_def.polarity_col = ""

        result = "Valid"
        # Not defined (missing columns in definition)
        if sph_def.type_ == "":
            result = "Not defined"
            return result

        if sph_def.type_ == "LINE":
            if sph_def.dip_col == "" or sph_def.azimuth_col == "":
                result = "Not defined"
                return result

        if sph_def.type_ == "PLANE":
            if (
                sph_def.dip_col == ""
                or sph_def.azimuth_col == ""
                or sph_def.polarity_col == ""
            ):
                result = "Not defined"
                return result

        ## fill polarity if a constant is selected
        if sph_def.polarity_col in self.polarity_mapping_item_delegate.other_options:
            raw_df[
                sph_def.polarity_col
            ] = self.polarity_mapping_item_delegate.other_options.index(
                sph_def.polarity_col
            )

        if not self._is_numeric_serie(raw_df[sph_def.dip_col]):
            result = "Wrong data type"
            return result

        if not self._is_numeric_serie(raw_df[sph_def.azimuth_col]):
            result = "Wrong data type"
            return result

        if sph_def.type_ == "PLANE":
            if not self._is_numeric_serie(raw_df[sph_def.polarity_col]):
                result = "Wrong data type"
                return result

        # Values outside range
        if (
            ~raw_df[sph_def.dip_col].astype(float).between(-90.0, 90.0)
        ).any() and sph_def.type_ == "LINE":
            result = "Wrong range"
            return result

        if (
            ~raw_df[sph_def.dip_col].astype(float).between(0.0, 90.0)
        ).any() and sph_def.type_ == "PLANE":
            result = "Wrong range"
            return result

        if (~raw_df[sph_def.azimuth_col].astype(float).between(0.0, 360.0)).any():
            result = "Wrong range"
            return result
        if sph_def.type_ == "PLANE":
            if (
                ~raw_df[sph_def.polarity_col].astype(float).isin([0.0, 1.0, 2.0])
            ).any():
                result = "Wrong range"
                return result

        return result

    @staticmethod
    def _is_numeric_serie(s: pandas.Series) -> bool:
        """
        Check if a pandas.Series can be converted to numeric values.
        Args:
            s : series
        Returns:
            True if convertible, else False
        """
        try:
            s = s.astype(float)
            return True
        except ValueError:

            return False

    def _add_wanted_columns(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """
        Add selected columns to dataframe if column not present

        Args:
            df: input dataframe

        Returns: dataframe with added columns if needed

        """
        all_column = self.column_mapping_model.get_column_definition()
        for mapping in all_column:
            col = mapping.column
            col_type = mapping.series_type
            if col not in df.columns:
                if col_type == AssaySeriesType.SPHERICAL:
                    df[col] = self._create_spherical_preview(col)

                else:
                    df[col] = None

        return df

    def _user_columns_conversion(self, df: pandas.DataFrame) -> pandas.DataFrame:
        """
        Convert dataframe columns with user specified conversion function

        Args:
            df: input dataframe

        Returns: dataframe with conversion

        """
        for col, conv in self._column_conversion.items():
            try:
                df[col] = conv(col, df)
            except BaseException as exc:
                QMessageBox.warning(self, "Invalid column conversion", str(exc))
                df[col] = None
        return df

    def crs(self) -> QgsCoordinateReferenceSystem:
        """
        Get user selected CRS

        Returns: user selected CRS

        """
        return self.mQgsProjectionSelectionWidget.crs()

    def set_crs(self, crs: QgsCoordinateReferenceSystem) -> None:
        """
        Define selected crs

        Args:
            crs: QgsCoordinateReferenceSystem
        """
        self.mQgsProjectionSelectionWidget.setCrs(crs)

    def enable_crs_selection(self, enable: bool) -> None:
        """
        Enable or disable CRS selection

        Args:
            enable: True for CRS selection enable, False for CRS selection disable
        """
        self.crs_label.setVisible(enable)
        self.mQgsProjectionSelectionWidget.setVisible(enable)

    def set_coordinates_columns(
        self,
        x_col: str = "",
        y_col: str = "",
        z_col: str = "",
        planned_x_col: str = "",
        planned_y_col: str = "",
        planned_z_col: str = "",
        dip_col: str = "",
        azimuth_col: str = "",
    ) -> None:
        """
        Define column names for coordinates data.
        """
        self._x_col = x_col
        self._y_col = y_col
        self._z_col = z_col
        self._planned_x_col = planned_x_col
        self._planned_y_col = planned_y_col
        self._planned_z_col = planned_z_col
        self._dip_col = dip_col
        self._azimuth_col = azimuth_col

    def enable_elevation_from_dtm(self, enable: bool):
        """
        Enable elevation calculation from selected DTM

        Args:
            enable: True for dtm display and elevation calculation
        """
        self.dtm_layer_combobox.setVisible(enable)
        self.dtm_label.setVisible(enable)

    def enable_date_format_selection(self, enable: bool) -> None:
        """
        Enable or disable date format selection

        Args:
            enable: True for data format enable, False for date format disable
        """
        self.dateformat_selection_widget.setVisible(enable)

    def set_date_format(self, date_format: str) -> None:
        """
        Define date format

        Args:
            date_format: wanted date format
        """
        self.dateformat_selection_widget.set_date_format(date_format)

    def set_time_format(self, time_format: str) -> None:
        """
        Define time format

        Args:
            time_format: wanted time format
        """
        self.dateformat_selection_widget.set_time_format(time_format)

    def _select_file(self) -> None:
        file_path, filter_used = QFileDialog.getOpenFileName(
            self, "Select file", "", "*.txt *.csv"
        )

        if file_path:
            self.filename_edit.lineEdit().setText(file_path)
            encoding = encoding_utils.predict_encoding(Path(file_path), 1)
            self.encoding_cbx.setCurrentText(encoding)
            self._read_data(file_path)
            self._update_table_and_fields()

    def _check_column_mapping(self) -> None:
        """
        Check if mapped column exists in csv file.
        If not, it is removed.
        """
        if self.df is None:
            return

        col_defs = self.get_column_definition()
        for col_def in col_defs:
            if col_def.mapping != "":
                if col_def.mapping not in self.df.columns:
                    col_def.mapping = ""

        self.set_column_definition(col_defs)

    def _update_uncertainty_table_model_columns(self) -> None:

        self.uncertainty_table_view.resizeColumnsToContents()

    def _creation_uncertainty_help_dialog(self) -> QDialog:
        help_dialog = QDialog(self)
        help_dialog.setWindowTitle("Uncertainty columns")
        layout = QHBoxLayout()
        label = QLabel()
        pixmap = QPixmap(str(DIR_PLUGIN_ROOT / "resources" / "images" / "boxplot.png"))
        pixmap = pixmap.scaled(
            600, 600, QtCore.Qt.KeepAspectRatio, QtCore.Qt.SmoothTransformation
        )
        label.setPixmap(pixmap)
        label.setScaledContents(True)
        layout.addWidget(label)
        help_dialog.setLayout(layout)
        return help_dialog

    def _creation_spherical_help_dialog(self) -> QDialog:
        help_dialog = QDialog(self)
        help_dialog.setWindowTitle("Spherical data definition")
        layout = QHBoxLayout()
        label = QLabel()
        pixmap = QPixmap(
            str(DIR_PLUGIN_ROOT / "resources" / "images" / "spherical_def.png")
        )
        pixmap = pixmap.scaled(
            1100, 600, QtCore.Qt.KeepAspectRatio, QtCore.Qt.SmoothTransformation
        )
        label.setPixmap(pixmap)
        label.setScaledContents(True)
        layout.addWidget(label)
        help_dialog.setLayout(layout)
        return help_dialog

    def _spherical_help_display(self) -> None:
        self.spherical_help_dialog.show()

    def _dtm_layer_changed(self) -> None:
        self.update_result_table()

    def _uncertainty_help_display(self) -> None:
        self.uncertainty_help_dialog.show()

    def _source_updated(self) -> None:
        """
        Read and store data as pandas.DataFrame.
        """
        file_path = self.get_file_path()
        self._read_data(file_path)
        self._update_table_and_fields()

    def _update_table_and_fields(self) -> None:
        self._update_sample_table_and_header_combo()
        self.update_result_table()

    def _update_sample_table_and_header_combo(self) -> None:

        if self.df is not None:
            array = self._get_sample_dataframe(100)
            self._update_table_content(self.sample_table_widget, array)
            headers = array.columns.to_list()
            self.column_mapping_item_delegate.available_columns = headers
            self.spherical_mapping_item_delegate.available_columns = headers
            self.polarity_mapping_item_delegate.available_columns = headers

            columns = self.column_mapping_model.get_column_definition()
            for mapping in columns:
                col = mapping.column
                # Automatic column mapping if header has same name as expected
                if col in headers and not mapping.mapping:
                    mapping.mapping = col

                # Check if column mapping is available
                if mapping.mapping not in headers:
                    mapping.mapping = ""

            self.column_mapping_model.dataChanged.disconnect(self.update_result_table)
            self.column_mapping_model.set_column_definition(columns)
            self.column_mapping_model.dataChanged.connect(self.update_result_table)

    def _create_category_policy(
        self, column_name: str, category: str = "", validation: str = ""
    ) -> CategoryPolicy:
        """
        Method creating default category name and validation policy.
        If these parameters are already edited by the user, we leave them as is.
        """
        existing_cat = [
            c.name
            for c in self._openlog_connection.get_categories_iface().get_available_categories_table()
        ]

        # if no name, propose default one
        category_name = f"{column_name}_cat" if category == "" else category

        # if no validation, propose default one
        if validation == "":
            val = CategoriesValidationType.APPEND
            if category_name in existing_cat:
                val = CategoriesValidationType.RESTRICT
            val = val.name
        else:
            val = validation

        return CategoryPolicy(
            column=column_name, category=category_name, validation=val
        )

    def _update_column_name(self, index, old_col, new_col) -> None:
        """
        Slot to rename dependant definitions after name change.
        Numerical : uncertainties and detection limits
        Spherical : spherical mapping
        """
        col_def = [
            col_def
            for col_def in self.get_column_definition()
            if col_def.column == old_col
        ]
        if len(col_def) == 0:
            return

        col_def = col_def[0]
        if col_def.series_type == AssaySeriesType.NUMERICAL:
            new_uncertainties = {}
            uncertainties = (
                self.mdl_assay_column_uncertainty.get_assay_column_uncertainty()
            )
            for k, v in uncertainties.items():
                if k == old_col:
                    new_uncertainties[new_col] = v
                else:
                    new_uncertainties[k] = v

            new_detection = {}
            detection = (
                self.mdl_assay_detection_limit.get_assay_column_detection_limit()
            )
            for k, v in detection.items():
                if k == old_col:
                    new_detection[new_col] = v
                else:
                    new_detection[k] = v

            # UNCERTAINTY UPDATE
            self.mdl_assay_column_uncertainty.dataChanged.disconnect(
                self.update_result_table
            )
            self.mdl_assay_column_uncertainty.set_assay_column_uncertainty(
                new_uncertainties
            )
            self.mdl_assay_column_uncertainty.dataChanged.connect(
                self.update_result_table
            )
            # DETECTION LIMIT UPDATE
            self.mdl_assay_detection_limit.dataChanged.disconnect(
                self.update_result_table
            )
            self.mdl_assay_detection_limit.set_assay_column_detection_limit(
                new_detection
            )
            self.mdl_assay_detection_limit.dataChanged.connect(self.update_result_table)

        elif col_def.series_type == AssaySeriesType.SPHERICAL:
            spherical = self.mdl_spherical_data.get_spherical_assay()
            new_spherical = {}
            for k, v in spherical.items():
                if k == old_col:
                    new_spherical[new_col] = v
                else:
                    new_spherical[k] = v

            self.mdl_spherical_data.dataChanged.disconnect(self.update_result_table)
            self.mdl_spherical_data.set_spherical_assay(new_spherical)
            self.mdl_spherical_data.dataChanged.connect(self.update_result_table)

    def update_result_table(self) -> None:

        file_path = self.filename_edit.lineEdit().text()
        columns = self.column_mapping_model.get_column_definition(False)
        # Check if datetime column is available
        self.enable_date_format_selection(
            any(x.series_type == AssaySeriesType.DATETIME for x in columns)
        )
        # Update column uncertainty table model
        current_column_uncertainty = (
            self.mdl_assay_column_uncertainty.get_assay_column_uncertainty()
        )

        current_detection_limit = (
            self.mdl_assay_detection_limit.get_assay_column_detection_limit()
        )
        # update spherical table model
        current_column_spherical = self.mdl_spherical_data.get_spherical_assay()

        current_column_categories = (
            self.mdl_assay_column_category.get_assay_column_categories()
        )
        column_uncertainty = {}
        column_detection = {}
        column_spherical = {}
        column_categorical = {}
        column_validation = {}
        for column in columns:
            col_name = column.column
            if column.series_type == AssaySeriesType.NUMERICAL and not column.fixed:
                # Get current uncertainty if defined
                if col_name not in current_column_uncertainty:
                    column_uncertainty[col_name] = AssayColumnUncertainty()
                else:
                    column_uncertainty[col_name] = current_column_uncertainty[col_name]

                # current detection limit
                if col_name not in current_detection_limit:
                    column_detection[col_name] = AssayColumnDetectionLimit()
                else:
                    column_detection[col_name] = current_detection_limit[col_name]

            if column.series_type == AssaySeriesType.CATEGORICAL:
                if col_name not in current_column_categories:
                    column_categorical[col_name] = ""
                else:
                    column_categorical[col_name] = current_column_categories[col_name]

                column_validation[
                    col_name
                ] = self.mdl_assay_column_category.get_assay_column_category_validation(
                    col_name
                )

            if column.series_type == AssaySeriesType.SPHERICAL:
                if col_name not in current_column_spherical:
                    column_spherical[col_name] = SphericalDefinition()
                else:
                    column_spherical[col_name] = current_column_spherical[col_name]

        categories = [
            self._create_category_policy(key, cat_name, column_validation[key])
            for key, cat_name in column_categorical.items()
        ]

        # SPHERICAL DEF UPDATE
        self.mdl_spherical_data.dataChanged.disconnect(self.update_result_table)
        self.mdl_spherical_data.set_spherical_assay(column_spherical)
        self.mdl_spherical_data.dataChanged.connect(self.update_result_table)
        self.gpx_spherical_data.setVisible(len(column_spherical) != 0)

        # if existing assay, already intialized by update_optionnal_widgets method
        if not self.already_created:
            # UNCERTAINTY UPDATE
            self.mdl_assay_column_uncertainty.dataChanged.disconnect(
                self.update_result_table
            )
            self.mdl_assay_column_uncertainty.set_assay_column_uncertainty(
                column_uncertainty
            )
            self.mdl_assay_column_uncertainty.dataChanged.connect(
                self.update_result_table
            )
            self.gpx_uncertainty.setVisible(len(column_uncertainty) != 0)

            # DETECTION LIMIT UPDATE
            self.mdl_assay_detection_limit.dataChanged.disconnect(
                self.update_result_table
            )
            self.mdl_assay_detection_limit.set_assay_column_detection_limit(
                column_detection
            )
            self.mdl_assay_detection_limit.dataChanged.connect(self.update_result_table)
            self.gpx_detection_limit.setVisible(len(column_detection) != 0)

        # CATEGORY UPDATE
        self.mdl_assay_column_category.dataChanged.disconnect(self.update_result_table)
        self.mdl_assay_column_category.set_assay_column_categories(categories)

        self.mdl_assay_column_category.dataChanged.connect(self.update_result_table)
        self.gpx_categories.setVisible(len(column_categorical) != 0)

        if self.df is not None:
            result = self.get_dataframe()
            self._update_table_content(self.result_table_widget, result, True)

    def _get_detection_limits_validity(self) -> dict:
        """
        Check if detection limits are valid.
        Return dict with [uncertainty columns] as key and tuple(validity:bool, message:str) as value.
        """
        df = self.get_dataframe()

        detection = self.get_assay_column_detection_limit()

        # check if mepped columns exists in data source, else set empty
        for k, det in detection.items():
            mapped_column = det.detection_min_col
            if mapped_column != "" and mapped_column not in df.columns:
                det.detection_min_col = ""
            mapped_column = det.detection_max_col
            if mapped_column != "" and mapped_column not in df.columns:
                det.detection_max_col = ""

        detection = {
            k: v for k, v in detection.items() if v.is_detection_limits_defined()
        }
        columns = self.get_column_definition()
        columns = [
            col for col in columns if col.series_type == AssaySeriesType.NUMERICAL
        ]
        display_names = [col.column for col in columns]
        # if existing dataset, columns are sanitized
        if self.already_created:
            db_names = [sanitize_sql_name(col.column) for col in columns]
        else:
            db_names = display_names.copy()

        result = {}
        # iterate over numerical columns
        for display_name, db_name in zip(display_names, db_names):
            if detection.get(db_name) is None:
                continue

            valid = True
            message = ""
            det = detection.get(db_name)
            if len(det.get_detection_columns()) == 2:
                lower_limit = (
                    df.loc[:, det.detection_min_col].replace("", np.nan).astype(float)
                )
                upper_limit = (
                    df.loc[:, det.detection_max_col].replace("", np.nan).astype(float)
                )

                if any(lower_limit > upper_limit):
                    valid = False
                    message = "Upper limit lower than lower limit"

            for det_column in det.get_detection_columns():
                result[det_column] = (valid, message)

        return result

    def _get_uncertainties_validity(self) -> dict:
        """
        Check if uncertainties are valid.
        Return dict with [uncertainty columns] as key and tuple(validity:bool, message:str) as value.
        """
        df = self.get_dataframe()
        uncertainties = self.get_assay_column_uncertainty()
        # check if mepped columns exists in data source, else set empty
        for k, uncert in uncertainties.items():
            mapped_column = uncert.upper_whisker_column
            if mapped_column != "" and mapped_column not in df.columns:
                uncert.upper_whisker_column = ""
            mapped_column = uncert.lower_whisker_column
            if mapped_column != "" and mapped_column not in df.columns:
                uncert.lower_whisker_column = ""
            mapped_column = uncert.upper_box_column
            if mapped_column != "" and mapped_column not in df.columns:
                uncert.upper_box_column = ""
            mapped_column = uncert.lower_box_column
            if mapped_column != "" and mapped_column not in df.columns:
                uncert.lower_box_column = ""

        uncertainties = {
            k: v
            for k, v in uncertainties.items()
            if v.get_uncertainty_type() != UncertaintyType.UNDEFINED
        }
        columns = self.get_column_definition()
        columns = [
            col for col in columns if col.series_type == AssaySeriesType.NUMERICAL
        ]
        display_names = [col.column for col in columns]
        # if existing dataset, columns are sanitized
        if self.already_created:
            db_names = [sanitize_sql_name(col.column) for col in columns]
        else:
            db_names = display_names.copy()

        result = {}
        # iterate over numerical columns
        for display_name, db_name in zip(display_names, db_names):
            if uncertainties.get(db_name) is None:
                continue

            valid = True
            message = ""
            uncert = uncertainties.get(db_name)
            if uncert.get_uncertainty_type() == UncertaintyType.ONE_COLUMN:
                interval = (
                    df.loc[:, uncert.upper_whisker_column]
                    .replace("", np.nan)
                    .astype(float)
                )
                if any(interval < 0):
                    valid = False
                    message = "Negative interval"
            elif uncert.get_uncertainty_type() == UncertaintyType.TWO_COLUMN:
                lower = (
                    df.loc[:, uncert.lower_whisker_column]
                    .replace("", np.nan)
                    .astype(float)
                )
                upper = (
                    df.loc[:, uncert.upper_whisker_column]
                    .replace("", np.nan)
                    .astype(float)
                )
                values = df.loc[:, display_name].replace("", np.nan).astype(float)
                if any(~((lower <= values) & (values <= upper))):
                    valid = False
                    message = "Δmin ≤ x ≤ Δmax not respected"
            elif uncert.get_uncertainty_type() == UncertaintyType.FOUR_COLUMN:
                lower_inner = (
                    df.loc[:, uncert.lower_whisker_column]
                    .replace("", np.nan)
                    .astype(float)
                )
                upper_inner = (
                    df.loc[:, uncert.upper_whisker_column]
                    .replace("", np.nan)
                    .astype(float)
                )
                lower_outer = (
                    df.loc[:, uncert.lower_box_column].replace("", np.nan).astype(float)
                )
                upper_outer = (
                    df.loc[:, uncert.upper_box_column].replace("", np.nan).astype(float)
                )
                values = df.loc[:, display_name].replace("", np.nan).astype(float)
                if any(
                    ~(
                        (lower_outer <= lower_inner)
                        & (lower_inner <= values)
                        & (values <= upper_inner)
                        & (upper_inner <= upper_outer)
                    )
                ):
                    valid = False
                    message = "Δmin ≤ δmin ≤ x ≤ δmax ≤ Δmax not respected"
            for uncert_columns in uncert.get_uncertainty_columns():
                result[uncert_columns] = (valid, message)

        return result

    def _update_table_content(
        self,
        table_widget: QTableWidget,
        array: pandas.DataFrame,
        display_errors: bool = False,
    ) -> None:
        # Dataframe columns are used as header
        headers = array.columns.astype(str).to_list()
        table_widget.setColumnCount(len(headers))
        table_widget.setHorizontalHeaderLabels(headers)

        spherical_data = self.mdl_spherical_data.get_spherical_assay().keys()
        # Insert all dataframe rows
        n_rows = min(100, array.shape[0])
        table_widget.setRowCount(n_rows)
        columns = array.columns
        self.red_flag = False

        # uncertainty validity
        uncert_validity = self._get_uncertainties_validity()
        # detection validity
        detection_validity = self._get_detection_limits_validity()

        mandatory_columns = [
            "HoleID*",
            "Hole ID*",
            "Depth*",
            "Time*",
            "Depth min.*",
            "Depth max.*",
            "Time min.*",
            "Time max.*",
            "Easting*",
            "Northing*",
            "Elevation*",
        ]
        for j in range(0, table_widget.columnCount()):
            i = 0
            for r in array.itertuples(index=False):
                if i > n_rows - 1:
                    break

                item = QTableWidgetItem(str(r[j]))
                table_widget.setItem(i, j, item)

                if r[j] is None:
                    if columns[j] in ["EOH"]:
                        item.setBackground(QColor("orange"))
                    else:
                        item.setBackground(QColor("yellow"))
                elif pandas.isna(r[j]):
                    item.setBackground(QColor("red"))

                # special case for spherical data
                if columns[j] in spherical_data:
                    if r[j] != "Valid":

                        item.setBackground(QColor("red"))
                    else:

                        item.setBackground(QColor("green"))

                # uncertainties validity
                if columns[j] in uncert_validity.keys():
                    validity, message = uncert_validity.get(columns[j])
                    if not validity:

                        item.setBackground(QColor("red"))
                        item.setText(message)

                # detection limits validity
                if columns[j] in detection_validity.keys():
                    validity, message = detection_validity.get(columns[j])
                    if not validity:
                        item.setBackground(QColor("red"))
                        item.setText(message)

                # check mandatory columns : hole, x,y,z, depths
                if columns[j] in mandatory_columns:

                    if pandas.isnull(r[j]):
                        item.setBackground(QColor("red"))

                i = i + 1

        # messages : key: column, value : [str]
        messages = {columns[i]: [] for i in range(table_widget.columnCount())}

        # check error in whole dataframe (use pandas for efficiency)
        for column in columns:
            s = array[column]
            # na (pandas.isna return True for None (missing value) and nan (conversion issue))
            nas = s[(s.apply(lambda x: x is not None)) & (s.isna())].index
            if len(nas):
                self.red_flag = True
                messages[column] += [f"row {i + 1} : cannot convert" for i in nas]
            # spherical
            if column in spherical_data:
                invalid = s[s != "Valid"].index
                if len(invalid):
                    self.red_flag = True
                    messages[column] += [
                        f"row {i + 1} : invalid spherical data" for i in invalid
                    ]

            # uncertainties
            if column in uncert_validity.keys():
                validity, message = uncert_validity.get(column)
                if not validity:
                    self.red_flag = True
                    messages[column].append("invalid uncertainties")

            # detection
            if column in detection_validity.keys():
                validity, message = detection_validity.get(column)
                if not validity:
                    self.red_flag = True
                    messages[column].append("invalid detection limits")

            # mandatory columns
            if column in mandatory_columns:
                nulls = s[s.apply(lambda x: x is None)].index
                if len(nulls):
                    self.red_flag = True
                    messages[column] += [f"row {i + 1} : missing value" for i in nulls]

            # PK duplicated
            if self.wizard_type == "collar":
                if column == "HoleID*":
                    duplicated = s.dropna().duplicated()
                    if any(duplicated):
                        self.red_flag = True
                        indexes = duplicated[duplicated].index
                        messages[column] += [
                            f"row {i + 1} : duplicated value" for i in indexes
                        ]

        # check column definition completeness
        self.red_flag = self.red_flag or not self._is_definition_valid()
        # emit red flag signal
        self.red_flag_signal.emit()

        if self.is_extended_assay:
            if self.UPDATE_STATUS_COL in array.columns:
                # Update item style
                for j in range(0, table_widget.columnCount()):
                    i = 0
                    for index, r in array.iterrows():
                        if i > n_rows - 1:
                            break
                        if r[self.UPDATE_STATUS_COL]:
                            font = table_widget.item(i, j).font()
                            font.setBold(True)
                            table_widget.item(i, j).setFont(font)
                        i = i + 1

                # Don't show update status col (last col). Must be reinitialized at each call because columns could be added
                n_col = table_widget.columnCount()
                for index in range(n_col):
                    if index == n_col - 1:
                        hide = True
                    else:
                        hide = False
                    table_widget.setColumnHidden(index, hide)

        if display_errors:
            self.messages = messages
            self._update_result_header_icons()

    def _update_result_header_icons(self) -> None:
        """
        Display an icon depending errors for each column.
        """
        headers = [
            self.result_table_widget.horizontalHeaderItem(i)
            for i in range(self.result_table_widget.columnCount())
        ]
        for header in headers:
            name = header.text()
            errors = self.messages.get(name)
            if errors is not None and len(errors) == 0:
                header.setIcon(QgsApplication.getThemeIcon("mIconSuccess.svg"))
            else:
                header.setIcon(QgsApplication.getThemeIcon("mIconDelete.svg"))

        for (
            i,
            header,
        ) in enumerate(headers):
            self.result_table_widget.setHorizontalHeaderItem(i, header)

        try:
            self.result_table_widget.horizontalHeader().sectionClicked.disconnect(
                self._display_error_messages
            )
        except Exception as e:
            pass
        self.result_table_widget.horizontalHeader().sectionClicked.connect(
            self._display_error_messages
        )

    def _display_error_messages(self, index):
        """
        Display error when clicking on header.
        """

        column_name = self.result_table_widget.horizontalHeaderItem(index).text()
        errors = self.messages.get(column_name)
        if errors is not None and len(errors) > 0:
            text = "\n".join(errors)
            dialog = BigTextDialog(text, self)
            dialog.exec_()

    def _is_definition_valid(self) -> bool:
        """
        Check if current assay definition is valid :
            - depths are defined
            - at least one measure column is defined
        If collar :
            - check if mandatory columns are mapped
        If survey:
            - check all columns are mapped
        Hole must be mapped in every case.
        """
        valid = True
        col_def = self.get_column_definition()
        if len(col_def) == 1:
            return False

        # hole
        valid = valid and col_def[0].mapping != ""
        if self.wizard_type == "assay":
            # depth
            valid = valid and col_def[1].mapping != ""
            if self.is_extended_assay:
                valid = valid and col_def[2].mapping != ""

            # columns
            if self.is_extended_assay:
                valid = valid and len(col_def) > 3
            else:
                valid = valid and len(col_def) > 2
        elif self.wizard_type == "collar":
            # x,y,z
            valid = (
                valid
                and col_def[1].mapping != ""
                and col_def[2].mapping != ""
                and col_def[3].mapping != ""
            )
        elif self.wizard_type == "survey":
            # dip, azimuth, depth
            valid = (
                valid
                and col_def[1].mapping != ""
                and col_def[2].mapping != ""
                and col_def[3].mapping != ""
            )

        return valid

    def _read_data(self, file_path: str):
        """
        Either read csv file, either convert QGIS layer into a pandas.DataFrame.
        """
        if str(file_path) == ".":
            return
        try:
            if str(file_path) in self.qgs_layers:
                file_path = str(file_path)

                self.delimiter_widget.hide()
                lyr = QgsProject.instance().mapLayersByName(file_path)[0]
                attributes_list = []

                start = self.header_row.value() - 1
                if self.header_row.value() == 1:
                    field_names = [field.name() for field in lyr.fields()]
                else:
                    field_names = list(lyr.getFeatures())[start].attributes()
                    field_names = [str(field) for field in field_names]

                for i, feature in enumerate(lyr.getFeatures()):
                    if i <= start:
                        continue
                    attributes = feature.attributes()
                    attributes_list.append(attributes)

                array = pandas.DataFrame(
                    attributes_list, columns=field_names, dtype=str
                )

                # replace NULL by None
                def replace_qvariant(value):
                    if isinstance(value, QtCore.QVariant):
                        return None
                    else:
                        return value

                array = array.applymap(replace_qvariant)

            elif str(file_path) in self.ags_tables.keys():
                array = self.ags_tables.get(str(file_path))
                # get assay definition to have domain and extent
                assay_def = None
                if self.wizard_type == "assay":
                    assay_def = self.parent().get_assay_definition()
                ags_columns_definitions = self.ags_reader.get_columns_definitions(
                    str(file_path), assay_def
                )

                self.mdl_assay_column_category.dataChanged.disconnect(
                    self.update_result_table
                )
                self.set_column_definition(ags_columns_definitions)
                self.mdl_assay_column_category.dataChanged.connect(
                    self.update_result_table
                )

                crs = self.ags_reader.get_crs()
                self.set_crs(crs)

            else:
                self.delimiter_widget.show()

                array = pandas.read_csv(
                    file_path,
                    delimiter=self.delimiter_widget.get_delimiter(),
                    skipinitialspace=True,
                    keep_default_na=False,
                    dtype=str,
                    engine="python",
                    quotechar=self.delimiter_widget.get_quote_char(),
                    escapechar=self.delimiter_widget.get_escape_char(),
                    encoding=self.encoding_cbx.currentText(),
                    skiprows=self.header_row.value() - 1,
                )
            self.df = array
            self._check_column_mapping()

        except BaseException as exc:
            array = pandas.DataFrame()
            msgBox = QMessageBox.warning(
                None,
                "Invalid file format definition.",
                f"Can't read input file. Details:\n{str(exc)}",
            )

    def _get_sample_dataframe(self, nrows: int = None) -> pandas.DataFrame:
        """
        Return a sample of raw dataframe.
        """
        if nrows is None:
            nrows = self.df.shape[0]
        array = self.df.iloc[:nrows, :]

        return array
