Skip to content

API

FM2PROF Runner Module.

This module provides the main execution engine for FM2PROF, a tool for converting 2D dflowFM model results to 1D cross-sections for hydraulic modelling.

FM2PROF (dflowFM to Profile) extracts cross-sectional data from 2D hydrodynamic model outputs and generates 1D model inputs. The main workflow includes:

  1. Initialisation: Load configuration files and validate input data -. Data Import: Read dflowfm map files and cross-section location files -. Classification: Assign 2D model points to regions and cross-sections
  2. Generation: Create cross-section geometries and roughness tables
  3. Finalisation: Write output files in various formats (D-Flow 1D, SOBEK 3, etc.)

Classes:

Name Description
InitializationError

Custom exception for initialisation failures.
Fm2ProfRunner

Main class that orchestrates the FM2PROF workflow.
Project

Python API wrapper for programmatic access to FM2PROF functionality.

The Project class is auto-imported when using the fm2prof package.

Example

Basic usage through the Project API:

from fm2prof import Project project = Project('config.ini') project.run()

Programmatic configuration:

project = Project() project.set_input_file('2DMapOutput', 'model_map.nc') project.set_input_file('CrossSectionLocationFile', 'crosssections.csv') project.set_output_directory('./output') project.run()

Note

This module requires FlowFM map files (NetCDF format) and cross-section location files as input. The output includes 1D model geometry and roughness data suitable for various hydraulic modelling software.

License

GPL-3.0-or-later AND LGPL-3.0-or-later

Fm2ProfRunner(ini_file_path='')

Bases: FM2ProfBase

Main class that executes all functionality.

Initialize the project.

ini_file_path (Path | str): path to configuration file.
Source code in fm2prof\fm2prof_runner.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(self, ini_file_path: Path | str = "") -> None:
    """Initialize the project.

    Args:
    ----
        ini_file_path (Path | str): path to configuration file.

    """
    self.fm_model_data: FmModelData = None
    self._output_files: OutputFiles = OutputFiles()

    self.set_logger(self.create_logger())

    ini_file_path = Path(ini_file_path)

    self.start_new_log_task("Loading configuration file")
    try:
        self.load_configuration(ini_file_path)
    except (ConfigurationFileError, FileNotFoundError) as e:
        self.set_logger_message(f"Exiting {e}", "error")
        return

    if not self.get_inifile().has_output_directory:
        self.set_logger_message(
            "Output directory must be set in configuration file",
            "error",
        )
        return

    # Add a log file
    self.set_logfile(
        output_dir=self.get_inifile().get_output_directory(),
        filename="fm2prof.log",
    )

    self.finish_log_task()
    # print header to log
    self._print_header()

    # Print configuration to log
    self.set_logger_message(self.get_inifile().print_configuration(), header=True)

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

load_configuration(ini_file_path)

Use this method to load a configuration file from path.

If no path is given, the default configuration is used.

ini_file_path (Path | str): path to configuration file
Source code in fm2prof\fm2prof_runner.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def load_configuration(self, ini_file_path: Path) -> None:
    """Use this method to load a configuration file from path.

    If no path is given, the default configuration is used.

    Args:
    ----
        ini_file_path (Path | str): path to configuration file

    """
    if not ini_file_path.is_file():
        self.set_logger_message("No ini file path given, using default configuration", "warning")
        ini_file_object = IniFile(logger=self.get_logger())
    else:
        ini_file_object = IniFile(ini_file_path, logger=self.get_logger())
    self.set_inifile(ini_file_object)

run(*, overwrite=False)

Execute FM2PROF routines.

Parameters:

Name Type Description Default
overwrite bool

if True, overwrites existing output. If False, exits if output detected.

 False

Returns:

Name Type Description
bool bool

True if run was successful, False if errors occurred.
Source code in fm2prof\fm2prof_runner.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def run(self, *, overwrite: bool = False) -> bool:
    """Execute FM2PROF routines.

    Args:
        overwrite (bool): if True, overwrites existing output. If False, exits if output detected.

    Returns:
        bool: True if run was successful, False if errors occurred.
    """
    if self.get_inifile() is None:
        self.set_logger_message(
            "No ini file was specified: the run cannot go further.",
            "Warning",
        )
        return False

    # Check for already existing output
    if self._output_exists() and not overwrite:
        self.set_logger_message(
            "Output already exists. Use overwrite option if you want to re-run the program",
            "warning",
        )
        return False

    # Run
    success = self._run_inifile()

    if not success:
        self.set_logger_message("Program finished with errors", "warning")
    else:
        self.set_logger_message("Program finished", "info")

    return success

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

InitializationError

Bases: Exception

Exception class for initialization errors.

Project(ini_file_path='')

Bases: Fm2ProfRunner

Provides the python API for running FM2PROF.

Instantiate by providing the path to a configuration file

Project('/path/to/config.ini')

Initialize the project.

ini_file_path (Path | str): path to configuration file.
Source code in fm2prof\fm2prof_runner.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(self, ini_file_path: Path | str = "") -> None:
    """Initialize the project.

    Args:
    ----
        ini_file_path (Path | str): path to configuration file.

    """
    self.fm_model_data: FmModelData = None
    self._output_files: OutputFiles = OutputFiles()

    self.set_logger(self.create_logger())

    ini_file_path = Path(ini_file_path)

    self.start_new_log_task("Loading configuration file")
    try:
        self.load_configuration(ini_file_path)
    except (ConfigurationFileError, FileNotFoundError) as e:
        self.set_logger_message(f"Exiting {e}", "error")
        return

    if not self.get_inifile().has_output_directory:
        self.set_logger_message(
            "Output directory must be set in configuration file",
            "error",
        )
        return

    # Add a log file
    self.set_logfile(
        output_dir=self.get_inifile().get_output_directory(),
        filename="fm2prof.log",
    )

    self.finish_log_task()
    # print header to log
    self._print_header()

    # Print configuration to log
    self.set_logger_message(self.get_inifile().print_configuration(), header=True)

output_files: Generator[Path, None, None] property

Get a generator object with the output files.

Generator[Path, None, None]: generator of output files.

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_input_file(name)

Use this method to retrieve the path to an input file.

name (str): case-insensitive key of the input file (e.g.'2dmapoutput')
Source code in fm2prof\fm2prof_runner.py 
1024
1025
1026
1027
1028
1029
1030
1031
1032
def get_input_file(self, name: str) -> str:
    """Use this method to retrieve the path to an input file.

    Args:
    ----
        name (str): case-insensitive key of the input file (e.g.'2dmapoutput')

    """
    return self.get_inifile().get_input_file(name)

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

get_output_directory()

Return the current output directory.

Source code in fm2prof\fm2prof_runner.py 
1048
1049
1050
def get_output_directory(self) -> Path:
    """Return the current output directory."""
    return self.get_inifile().get_output_directory()

get_parameter(name)

Use this method to get the value of a parameter.

name (str): name of the parameter (case insensitive)

(str | float): The current value of the parameter
Source code in fm2prof\fm2prof_runner.py 
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
def get_parameter(self, name: str) -> str | float:
    """Use this method to get the value of a parameter.

    Args:
    ----
        name (str): name of the parameter (case insensitive)

    Returns:
    -------
        (str | float): The current value of the parameter

    """
    return self.get_inifile().get_parameter(name)

load_configuration(ini_file_path)

Use this method to load a configuration file from path.

If no path is given, the default configuration is used.

ini_file_path (Path | str): path to configuration file
Source code in fm2prof\fm2prof_runner.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def load_configuration(self, ini_file_path: Path) -> None:
    """Use this method to load a configuration file from path.

    If no path is given, the default configuration is used.

    Args:
    ----
        ini_file_path (Path | str): path to configuration file

    """
    if not ini_file_path.is_file():
        self.set_logger_message("No ini file path given, using default configuration", "warning")
        ini_file_object = IniFile(logger=self.get_logger())
    else:
        ini_file_object = IniFile(ini_file_path, logger=self.get_logger())
    self.set_inifile(ini_file_object)

print_configuration()

Use this method to obtain string representation of the configuration.

Use this string to write to file, e.g.:

>> with open('EmptyProject.ini', 'w') as f:
>>     f.write(project.print_configuration())

(str): string representation of the configuration
Source code in fm2prof\fm2prof_runner.py 
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
def print_configuration(self) -> str:
    """Use this method to obtain string representation of the configuration.

    Use this string to write to file, e.g.:

        >> with open('EmptyProject.ini', 'w') as f:
        >>     f.write(project.print_configuration())

    Returns:
    -------
        (str): string representation of the configuration

    """
    return self.get_inifile().print_configuration()

run(*, overwrite=False)

Execute FM2PROF routines.

Parameters:

Name Type Description Default
overwrite bool

if True, overwrites existing output. If False, exits if output detected.

 False

Returns:

Name Type Description
bool bool

True if run was successful, False if errors occurred.
Source code in fm2prof\fm2prof_runner.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def run(self, *, overwrite: bool = False) -> bool:
    """Execute FM2PROF routines.

    Args:
        overwrite (bool): if True, overwrites existing output. If False, exits if output detected.

    Returns:
        bool: True if run was successful, False if errors occurred.
    """
    if self.get_inifile() is None:
        self.set_logger_message(
            "No ini file was specified: the run cannot go further.",
            "Warning",
        )
        return False

    # Check for already existing output
    if self._output_exists() and not overwrite:
        self.set_logger_message(
            "Output already exists. Use overwrite option if you want to re-run the program",
            "warning",
        )
        return False

    # Run
    success = self._run_inifile()

    if not success:
        self.set_logger_message("Program finished with errors", "warning")
    else:
        self.set_logger_message("Program finished", "info")

    return success

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_input_file(name, value)

Use this method to set the path to an input file.

name: name of the input file in the configuration (case insensitive).

value: path to the inputfile
Source code in fm2prof\fm2prof_runner.py 
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
def set_input_file(self, name: str, value: str | float) -> None:
    """Use this method to set the path to an input file.

    Args:
    ----
        name: name of the input file in the configuration (case insensitive).

        value: path to the inputfile

    """
    return self.get_inifile().set_input_file(name, value)

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

set_output_directory(path)

Use this method to set the output directory.

.. warning:: calling this function will also create the output directory, if it does not already exists!

path (path | str): path to the output path
Source code in fm2prof\fm2prof_runner.py 
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
def set_output_directory(self, path: str | Path) -> None:
    """Use this method to set the output directory.

    .. warning::
        calling this function will also create the output directory,
        if it does not already exists!

    Args:
    ----
        path (path | str): path to the output path

    """
    self.get_inifile().set_output_directory(path)

set_parameter(name, value)

Use this method to set the value of a parameter.

name (str): name of the parameter (case insensitive).

value (str | float): value of the parameter.
An error will be given if the value has the wrong type (e.g. string if int was expected).
Source code in fm2prof\fm2prof_runner.py 
985
986
987
988
989
990
991
992
993
994
995
996
def set_parameter(self, name: str, value: str | float) -> None:
    """Use this method to set the value of a parameter.

    Args:
    ----
        name (str): name of the parameter (case insensitive).

        value (str | float): value of the parameter.
        An error will be given if the value has the wrong type (e.g. string if int was expected).

    """
    self.get_inifile().set_parameter(name, value)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

Region and Section Polygon File Module.

This module provides functionality for handling polygon files used in FM2PROF for spatial classification of 2D model data into regions and sections.

The module supports GeoJSON format polygon files and provides efficient spatial indexing and point-in-polygon classification algorithms. It handles two main types of polygon files:

  1. Region Polygons: Define geographical regions for grouping cross-sections
  2. Section Polygons: Define hydraulic sections (main channel, floodplains)
Key Features
  • GeoJSON file parsing and validation
  • Spatial indexing using MeshKernel for efficient point classification
  • Overlap detection and validation

Classes:

Name Description
Polygon

Named tuple representing a polygon with geometry and properties.
PolygonFile

Base class for polygon file handling with classification methods.
RegionPolygonFile

Specialised class for region polygon files.
SectionPolygonFile

Specialised class for section polygon files with hydraulic validation.
Note

Polygon files must be in GeoJSON format with specific property requirements: - Region polygons: require 'name' and 'id' properties - Section polygons: require 'name' and 'section' properties ('main', 'floodplain1', 'floodplain2')

GridPointsInPolygonResults

Bases: NamedTuple

Named tuple for grid points in polygon results.

MultiPolygon(logger)

Bases: FM2ProfBase

MultiPolygon file class.

This class handles MultiPolygon files used in FM2PROF for spatial classification of 2D model data into regions and sections. It supports GeoJSON format polygon files and provides methods for reading, validating, and classifying points within the polygons.

We use a base MultiPolygon class to leverage MeshKernel and Shapely functionality within a common framework. E.g. MultiPolygon.as_meshkernel() outputs a MeshKernel GeometryList object that can be used for spatial classification, while MultiPolygon.as_shapely() outputs a list of Shapely Polygon objects for geometric operations.

Not all geojson geometry types are supported:

  • Only 'Polygon' and 'MultiPolygon' are supported.
  • Polygons with holes are not supported.
  • MultiPolygons with multiple polygons are not supported.
  • Properties must contain a 'name' key-word.
  • Properties must be unique, otherwise overlap checking will produce bugs.
  • SectionPolygon properties must contain a 'section' key-word.
  • RegionPolygon properties must contain a 'region' key-word.

Instantiate a MultiPolygon object.

Source code in fm2prof\polygon_file.py 
117
118
119
120
121
def __init__(self, logger: Logger) -> None:
    """Instantiate a MultiPolygon object."""
    self.set_logger(logger)
    self._polygons = []
    self.undefined = "undefined"

polygons: list[Polygon] property writable

Polygons.

as_meshkernel()

Convert polygons to MeshKernel GeometryList.

Note

MeshKernel GeometryList supports multiple polygons, separated by some int (default -999). However, to keep track of polygon properties (e.g. name), we create a list of single polygon GeometryList objects.

Returns:

Name Type Description
GeometryList list[GeometryList]

MeshKernel GeometryList object containing all polygons.
Source code in fm2prof\polygon_file.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def as_meshkernel(self) -> list[GeometryList]:
    """Convert polygons to MeshKernel GeometryList.

    Note:
        MeshKernel GeometryList supports multiple polygons,
        separated by some int (default -999). However,
        to keep track of polygon properties (e.g. name),
        we create a list of single polygon GeometryList objects.

    Returns:
        GeometryList: MeshKernel GeometryList object containing all polygons.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [GeometryList(x_coordinates=polygon.x, y_coordinates=polygon.y) for polygon in self.polygons]

as_shapely()

Convert polygons to list of Shapely Polygon objects.

Returns:

Type Description
list[Polygon]

list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
Source code in fm2prof\polygon_file.py 
162
163
164
165
166
167
168
169
170
171
172
def as_shapely(self) -> list[shapely.geometry.Polygon]:
    """Convert polygons to list of Shapely Polygon objects.

    Returns:
        list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [shapely.Polygon(polygon.coordinates) for polygon in self.polygons]

check_overlap()

Check if polygons overlap and log a warning if they do.

Source code in fm2prof\polygon_file.py 
332
333
334
335
336
337
338
339
340
341
342
343
def check_overlap(self) -> None:
    """Check if polygons overlap and log a warning if they do."""
    for i, poly1 in enumerate(self.as_shapely()):
        for j, poly2 in enumerate(self.as_shapely()):
            if i == j:
                # polygon will obviously overlap with itself
                continue
            if poly1.intersects(poly2):
                self.set_logger_message(
                   f"{self.polygons[i].properties.get('name')} overlaps {self.polygons[j].properties.get('name')}.",
                   level="warning",
                )

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

from_file(file_path)

Read data from geojson file.

Parameters:

Name Type Description Default
file_path Path | str

path to geojson file

 required
Source code in fm2prof\polygon_file.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def from_file(self, file_path: Path | str) -> None:
    """Read data from geojson file.

    Args:
        file_path (Path | str): path to geojson file
    """
    self._validate_extension(file_path)

    if Path(file_path).exists() is False:
        err_msg = "Polygon file does not exist"
        raise FileNotFoundError(err_msg)

    with Path(file_path).open("r") as geojson_file:
        try:
            geojson_data = json.load(geojson_file).get("features")
        except json.JSONDecodeError as e:
            err_msg = f"Error decoding JSON from {file_path}: {e}"
            raise PolygonError(err_msg) from e

    if not geojson_data:
        err_msg = "Polygon file has no features"
        raise PolygonError(err_msg)

    self.polygons = self._polygons_from_geojson_data(geojson_data)

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_gridpoints_in_polygon(res_file, *, property_name, force_cache_invalidation=False)

Method to get faces and edges in region.

This method performs caching of the in-polygon classification results to avoid recalculating if the region file has not changed. The cache is invalidated if the region file is modified or if force_cache_invalidation is set to True.

Parameters:

Name Type Description Default
res_file str | Path

path to result (map) netcdf file.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required
force_cache_invalidation bool

Force cache invalidation even if region file has not changed.

 False

Returns:

Type Description
GridPointsInPolygonResults

GridPointsInPolygonResults
Source code in fm2prof\polygon_file.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def get_gridpoints_in_polygon(self,
    res_file: str | Path,
    *,
    property_name: Literal["region", "section"],
    force_cache_invalidation: bool = False) -> GridPointsInPolygonResults:
    """Method to get faces and edges in region.

    This method performs caching of the in-polygon classification results
    to avoid recalculating if the region file has not changed. The cache
    is invalidated if the region file is modified or if `force_cache_invalidation`
    is set to `True`.

    Args:
        res_file (str | Path): path to result (map) netcdf file.
        property_name (Literal["region", "section"]): Property to use for classification.
        force_cache_invalidation (bool): Force cache invalidation even if region file has not changed.

    Returns:
        GridPointsInPolygonResults
    """
    # get metadata of file path to determine if the in-polygon classification needs
    # to be rerun
    meta = {"last_modified": res_file.stat().st_mtime,
            "file_size": res_file.stat().st_size}

    # check if cache file exists and is valid
    cache_file = Path(res_file).with_suffix(f".{property_name}_cache.json")
    if cache_file.exists() and not force_cache_invalidation:
        # read cache file
        self.set_logger_message("Found cached regions, reading...", level="info")
        meta_cache, faces_in_polygon, edges_in_polygon = self._load_cache(cache_file)
        if meta == meta_cache:
            self.set_logger_message("Using cached regions", level="info")
            return GridPointsInPolygonResults(faces_in_polygon=faces_in_polygon, edges_in_polygon=edges_in_polygon)
        self.set_logger_message("Cached regions are stale", level="warning")
        faces_in_polygon = self.meshkernel_inpolygon(res_file, dtype="face", property_name=property_name)
        edges_in_polygon = self.meshkernel_inpolygon(res_file, dtype="edge", property_name=property_name)
        self._write_cache(cache_file, meta, faces_in_polygon, edges_in_polygon)
    elif cache_file.exists() and force_cache_invalidation:
        self.set_logger_message("Forcing recalculating region cache", level="info")
        cache_file.unlink()
        faces_in_polygon = self.meshkernel_inpolygon(res_file, dtype="face", property_name=property_name)
        edges_in_polygon = self.meshkernel_inpolygon(res_file, dtype="edge", property_name=property_name)
        self._write_cache(cache_file, meta, faces_in_polygon, edges_in_polygon)
    elif not cache_file.exists():
        faces_in_polygon = self.meshkernel_inpolygon(res_file, dtype="face", property_name=property_name)
        edges_in_polygon = self.meshkernel_inpolygon(res_file, dtype="edge", property_name=property_name)
        self._write_cache(cache_file, meta, faces_in_polygon, edges_in_polygon)

    return GridPointsInPolygonResults(faces_in_polygon=faces_in_polygon, edges_in_polygon=edges_in_polygon)

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

get_points_in_polygon(points, property_name)

Method to determine in which polygon input points are.

Warning

This method is not applicable for large number of points. Only use for small number of points (e.g. cross-section locations).

Parameters:

Name Type Description Default
points ndarray

Array of shape (n_points, 2) containing x,y coordinates of points to classify.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

list[str]: List of polygon names in which the points are located. If a point is not located in any polygon, it is classified as 'undefined'.
Source code in fm2prof\polygon_file.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_points_in_polygon(self, points: np.ndarray, property_name: Literal["region", "section"]) -> list[str]:
    """Method to determine in which polygon input points are.

    Warning:
        This method is not applicable for large number of points.
        Only use for small number of points (e.g. cross-section locations).

    Args:
        points (np.ndarray): Array of shape (n_points, 2) containing x,y coordinates of points to classify.
        property_name (Literal['region', 'section']): Property to use for classification.

    Returns:
        list[str]: List of polygon names in which the points are located. If a point is not located
                   in any polygon, it is classified as 'undefined'.
    """
    # Convert to shapely point
    points = [shapely.Point(xy) for xy in points]
    points_regions = [self.undefined] * len(points)

    # Assign point to region
    for i, point in enumerate(points):
        for j, polygon in enumerate(self.as_shapely()):
            if point.within(polygon):
                points_regions[i] = self.polygons[j].properties.get(property_name)
                break

    return points_regions

meshkernel_inpolygon(res_file, dtype, property_name)

Get grid points in polygon.

Parameters:

Name Type Description Default
res_file str | Path

Path to result (map) netcdf file.

 required
dtype Literal['face', 'edge', 'node']

Type of grid points to retrieve.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
Source code in fm2prof\polygon_file.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def meshkernel_inpolygon(self,
    res_file: str | Path,  # path to result (map) netcdf file
    dtype: Literal["face", "edge", "node"],
    property_name: Literal["region", "section"],
    ) -> list[str]:
    """Get grid points in polygon.

    Args:
        res_file (str | Path): Path to result (map) netcdf file.
        dtype (Literal["face", "edge", "node"]): Type of grid points to retrieve.
        property_name (Literal["region", "section"]): Property to use for classification.

    Returns:
        pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
    """
    # Step 1
    # Construct Meshkernel grid
    mk = MeshKernel(projection=ProjectionType.CARTESIAN)
    mesh2d_input = mk.mesh2d_get()

    fmdata = FMDataImporter(res_file)
    mesh2d_input.node_x = fmdata.get_variable("mesh2d_node_x")
    mesh2d_input.node_y = fmdata.get_variable("mesh2d_node_y")
    mesh2d_input.edge_nodes = fmdata.get_variable("mesh2d_edge_nodes").flatten() - 1

    mk.mesh2d_set(mesh2d_input)

    # Step 2: perform point-in-polygon classification
    nodes_in_polygon: list[str] = [self.undefined] * len(mesh2d_input.node_x)  # default to undefined

    for i, mk_polygon in enumerate(self.as_meshkernel()):
        self.set_logger_message(
            f"  | Classifying {dtype}s in polygon {self.polygons[i].properties.get('name')} "
            f"({i+1}/{len(self.polygons)})",
            level="info")
        indices: list[float] = mk.mesh2d_get_nodes_in_polygons(mk_polygon, inside=True).tolist()
        for j in indices:
            nodes_in_polygon[j] = self.polygons[i].properties.get(property_name)

    # Step 3: map to faces or edges if needed
    if dtype == "node":
        output = nodes_in_polygon
    elif dtype == "face":
        face_map: np.ndarray = fmdata.get_variable("mesh2d_face_nodes").T[0] -1
        node_to_face: list[str] = [self.undefined] * len(face_map)
        for face_index, map_index in enumerate(face_map.tolist()):
            node_to_face[int(face_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_face
    elif dtype == "edge":
        # only internal edges!
        internal_edges = fmdata.get_variable("mesh2d_edge_type")[:] == 1
        edge_map: np.ndarray = fmdata.get_variable("mesh2d_edge_nodes").T[0] -1
        edge_map = edge_map[internal_edges]
        node_to_edge: list[str] = [self.undefined] * len(edge_map)
        for edge_index, map_index in enumerate(edge_map.tolist()):
            node_to_edge[int(edge_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_edge

    return output

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

Polygon(coordinates, properties)

Polygon class.

This class represents a single polygon with its geometry and properties. It is used by the MultiPolygon class to store individual polygons.

Attributes:

Name Type Description
coordinates list[list[float]]

List of [x, y] coordinates defining the polygon.
properties dict

Dictionary of properties associated with the polygon.

Instantiate a Polygon object.

Parameters:

Name Type Description Default
coordinates list[list[float]]

List of [x, y] coordinates defining the polygon.

 required
properties dict

Dictionary of properties associated with the polygon.

 required
Source code in fm2prof\polygon_file.py 
72
73
74
75
76
77
78
79
80
def __init__(self, coordinates: list[list[float]], properties: dict) -> None:
    """Instantiate a Polygon object.

    Args:
        coordinates (list[list[float]]): List of [x, y] coordinates defining the polygon.
        properties (dict): Dictionary of properties associated with the polygon.
    """
    self.coordinates = coordinates
    self.properties = properties

x: list[float] property

X coordinates of the polygon.

y: list[float] property

Y coordinates of the polygon.

PolygonError(message)

Bases: Exception

Custom exception for polygon errors.

Initialize PolygonError with a message.

Source code in fm2prof\polygon_file.py 
50
51
52
53
def __init__(self, message: str) -> None:
    """Initialize PolygonError with a message."""
    self.message = message
    super().__init__(self.message)

RegionPolygon(region_file_path, logger, default_value='undefined')

Bases: MultiPolygon

RegionPolygonFile class.

Instantiate a RegionPolygonFile object.

Parameters:

Name Type Description Default
region_file_path str | Path

path to region polygon file.

 required
logger Logger

logger

 required
default_value str

default region name to use in cells not covered by a region polygon

 'undefined'
Source code in fm2prof\polygon_file.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
def __init__(self, region_file_path: str | Path, logger: Logger, default_value: str="undefined") -> None:
    """Instantiate a RegionPolygonFile object.

    Args:
        region_file_path (str | Path): path to region polygon file.
        logger (Logger): logger
        default_value (str): default region name to use in cells not covered by a region polygon
    """
    super().__init__(logger)
    try:
        self.from_file(region_file_path)
    except TypeError as e:
        err_msg = f"Potentially invalid geojson file: {e}"
        self.set_logger_message(err_msg, level="error")
        raise PolygonError(err_msg) from e

    self.undefined = default_value

polygons: list[Polygon] property writable

Polygons.

regions: list[Polygon] property

Region polygons.

as_meshkernel()

Convert polygons to MeshKernel GeometryList.

Note

MeshKernel GeometryList supports multiple polygons, separated by some int (default -999). However, to keep track of polygon properties (e.g. name), we create a list of single polygon GeometryList objects.

Returns:

Name Type Description
GeometryList list[GeometryList]

MeshKernel GeometryList object containing all polygons.
Source code in fm2prof\polygon_file.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def as_meshkernel(self) -> list[GeometryList]:
    """Convert polygons to MeshKernel GeometryList.

    Note:
        MeshKernel GeometryList supports multiple polygons,
        separated by some int (default -999). However,
        to keep track of polygon properties (e.g. name),
        we create a list of single polygon GeometryList objects.

    Returns:
        GeometryList: MeshKernel GeometryList object containing all polygons.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [GeometryList(x_coordinates=polygon.x, y_coordinates=polygon.y) for polygon in self.polygons]

as_shapely()

Convert polygons to list of Shapely Polygon objects.

Returns:

Type Description
list[Polygon]

list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
Source code in fm2prof\polygon_file.py 
162
163
164
165
166
167
168
169
170
171
172
def as_shapely(self) -> list[shapely.geometry.Polygon]:
    """Convert polygons to list of Shapely Polygon objects.

    Returns:
        list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [shapely.Polygon(polygon.coordinates) for polygon in self.polygons]

check_overlap()

Check if polygons overlap and log a warning if they do.

Source code in fm2prof\polygon_file.py 
332
333
334
335
336
337
338
339
340
341
342
343
def check_overlap(self) -> None:
    """Check if polygons overlap and log a warning if they do."""
    for i, poly1 in enumerate(self.as_shapely()):
        for j, poly2 in enumerate(self.as_shapely()):
            if i == j:
                # polygon will obviously overlap with itself
                continue
            if poly1.intersects(poly2):
                self.set_logger_message(
                   f"{self.polygons[i].properties.get('name')} overlaps {self.polygons[j].properties.get('name')}.",
                   level="warning",
                )

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

from_file(file_path)

Read geojson file and performs validation.

Parameters:

Name Type Description Default
file_path Path | str

region file path

 required
Source code in fm2prof\polygon_file.py 
434
435
436
437
438
439
440
441
442
def from_file(self, file_path: Path | str) -> None:
    """Read geojson file and performs validation.

    Args:
        file_path (Path | str): region file path

    """
    super().from_file(file_path)
    self._validate_regions()

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_gridpoints_in_polygon(res_file, *, property_name='region', force_cache_invalidation=False)

Method to get faces and edges in region.

This method performs caching of the in-polygon classification results to avoid recalculating if the region file has not changed. The cache is invalidated if the region file is modified or if force_cache_invalidation is set to True.

Parameters:

Name Type Description Default
res_file str | Path

path to result (map) netcdf file.

 required
property_name Literal['region', 'section']

Property to use for classification. Defaults to region

 'region'
force_cache_invalidation bool

Force cache invalidation even if region file has not changed.

 False

Returns:

Type Description
GridPointsInPolygonResults

GridPointsInPolygonResults
Source code in fm2prof\polygon_file.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
def get_gridpoints_in_polygon(self,
    res_file: str | Path,
    *,
    property_name: Literal["region", "section"]="region",
    force_cache_invalidation:bool=False) -> GridPointsInPolygonResults:
    """Method to get faces and edges in region.

    This method performs caching of the in-polygon classification results
    to avoid recalculating if the region file has not changed. The cache
    is invalidated if the region file is modified or if `force_cache_invalidation`
    is set to `True`.

    Args:
        res_file (str | Path): path to result (map) netcdf file.
        property_name (Literal["region", "section"]): Property to use for classification. Defaults to region
        force_cache_invalidation (bool): Force cache invalidation even if region file has not changed.

    Returns:
        GridPointsInPolygonResults
    """
    return super().get_gridpoints_in_polygon(res_file,
                                          property_name=property_name,
                                          force_cache_invalidation=force_cache_invalidation)

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

get_points_in_polygon(points, property_name)

Method to determine in which polygon input points are.

Warning

This method is not applicable for large number of points. Only use for small number of points (e.g. cross-section locations).

Parameters:

Name Type Description Default
points ndarray

Array of shape (n_points, 2) containing x,y coordinates of points to classify.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

list[str]: List of polygon names in which the points are located. If a point is not located in any polygon, it is classified as 'undefined'.
Source code in fm2prof\polygon_file.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_points_in_polygon(self, points: np.ndarray, property_name: Literal["region", "section"]) -> list[str]:
    """Method to determine in which polygon input points are.

    Warning:
        This method is not applicable for large number of points.
        Only use for small number of points (e.g. cross-section locations).

    Args:
        points (np.ndarray): Array of shape (n_points, 2) containing x,y coordinates of points to classify.
        property_name (Literal['region', 'section']): Property to use for classification.

    Returns:
        list[str]: List of polygon names in which the points are located. If a point is not located
                   in any polygon, it is classified as 'undefined'.
    """
    # Convert to shapely point
    points = [shapely.Point(xy) for xy in points]
    points_regions = [self.undefined] * len(points)

    # Assign point to region
    for i, point in enumerate(points):
        for j, polygon in enumerate(self.as_shapely()):
            if point.within(polygon):
                points_regions[i] = self.polygons[j].properties.get(property_name)
                break

    return points_regions

meshkernel_inpolygon(res_file, dtype, property_name)

Get grid points in polygon.

Parameters:

Name Type Description Default
res_file str | Path

Path to result (map) netcdf file.

 required
dtype Literal['face', 'edge', 'node']

Type of grid points to retrieve.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
Source code in fm2prof\polygon_file.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def meshkernel_inpolygon(self,
    res_file: str | Path,  # path to result (map) netcdf file
    dtype: Literal["face", "edge", "node"],
    property_name: Literal["region", "section"],
    ) -> list[str]:
    """Get grid points in polygon.

    Args:
        res_file (str | Path): Path to result (map) netcdf file.
        dtype (Literal["face", "edge", "node"]): Type of grid points to retrieve.
        property_name (Literal["region", "section"]): Property to use for classification.

    Returns:
        pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
    """
    # Step 1
    # Construct Meshkernel grid
    mk = MeshKernel(projection=ProjectionType.CARTESIAN)
    mesh2d_input = mk.mesh2d_get()

    fmdata = FMDataImporter(res_file)
    mesh2d_input.node_x = fmdata.get_variable("mesh2d_node_x")
    mesh2d_input.node_y = fmdata.get_variable("mesh2d_node_y")
    mesh2d_input.edge_nodes = fmdata.get_variable("mesh2d_edge_nodes").flatten() - 1

    mk.mesh2d_set(mesh2d_input)

    # Step 2: perform point-in-polygon classification
    nodes_in_polygon: list[str] = [self.undefined] * len(mesh2d_input.node_x)  # default to undefined

    for i, mk_polygon in enumerate(self.as_meshkernel()):
        self.set_logger_message(
            f"  | Classifying {dtype}s in polygon {self.polygons[i].properties.get('name')} "
            f"({i+1}/{len(self.polygons)})",
            level="info")
        indices: list[float] = mk.mesh2d_get_nodes_in_polygons(mk_polygon, inside=True).tolist()
        for j in indices:
            nodes_in_polygon[j] = self.polygons[i].properties.get(property_name)

    # Step 3: map to faces or edges if needed
    if dtype == "node":
        output = nodes_in_polygon
    elif dtype == "face":
        face_map: np.ndarray = fmdata.get_variable("mesh2d_face_nodes").T[0] -1
        node_to_face: list[str] = [self.undefined] * len(face_map)
        for face_index, map_index in enumerate(face_map.tolist()):
            node_to_face[int(face_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_face
    elif dtype == "edge":
        # only internal edges!
        internal_edges = fmdata.get_variable("mesh2d_edge_type")[:] == 1
        edge_map: np.ndarray = fmdata.get_variable("mesh2d_edge_nodes").T[0] -1
        edge_map = edge_map[internal_edges]
        node_to_edge: list[str] = [self.undefined] * len(edge_map)
        for edge_index, map_index in enumerate(edge_map.tolist()):
            node_to_edge[int(edge_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_edge

    return output

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

SectionPolygon(section_file_path, logger, default_value='floodplain1')

Bases: MultiPolygon

SectionPolygonFile class.

Instantiate a SectionPolygonFile object.

Parameters:

Name Type Description Default
section_file_path str | Path

path to section polygon file.

 required
logger Logger

logger

 required
default_value str

default section name to use in cells not covered by a section polygon.

 'floodplain1'
Source code in fm2prof\polygon_file.py 
482
483
484
485
486
487
488
489
490
491
492
493
def __init__(self, section_file_path: str | Path, logger: Logger, default_value: str="floodplain1") -> None:
    """Instantiate a SectionPolygonFile object.

    Args:
        section_file_path (str | Path): path to section polygon file.
        logger (Logger): logger
        default_value (str): default section name to use in cells not covered by a section polygon.

    """
    super().__init__(logger)
    self.from_file(section_file_path)
    self.undefined = default_value

polygons: list[Polygon] property writable

Polygons.

sections: list[Polygon] property

Section polygons.

as_meshkernel()

Convert polygons to MeshKernel GeometryList.

Note

MeshKernel GeometryList supports multiple polygons, separated by some int (default -999). However, to keep track of polygon properties (e.g. name), we create a list of single polygon GeometryList objects.

Returns:

Name Type Description
GeometryList list[GeometryList]

MeshKernel GeometryList object containing all polygons.
Source code in fm2prof\polygon_file.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def as_meshkernel(self) -> list[GeometryList]:
    """Convert polygons to MeshKernel GeometryList.

    Note:
        MeshKernel GeometryList supports multiple polygons,
        separated by some int (default -999). However,
        to keep track of polygon properties (e.g. name),
        we create a list of single polygon GeometryList objects.

    Returns:
        GeometryList: MeshKernel GeometryList object containing all polygons.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [GeometryList(x_coordinates=polygon.x, y_coordinates=polygon.y) for polygon in self.polygons]

as_shapely()

Convert polygons to list of Shapely Polygon objects.

Returns:

Type Description
list[Polygon]

list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
Source code in fm2prof\polygon_file.py 
162
163
164
165
166
167
168
169
170
171
172
def as_shapely(self) -> list[shapely.geometry.Polygon]:
    """Convert polygons to list of Shapely Polygon objects.

    Returns:
        list[shapely.geometry.Polygon]: List of Shapely Polygon objects.
    """
    if not self.polygons:
        err_msg = "No polygons defined"
        raise ValueError(err_msg)

    return [shapely.Polygon(polygon.coordinates) for polygon in self.polygons]

check_overlap()

Check if polygons overlap and log a warning if they do.

Source code in fm2prof\polygon_file.py 
332
333
334
335
336
337
338
339
340
341
342
343
def check_overlap(self) -> None:
    """Check if polygons overlap and log a warning if they do."""
    for i, poly1 in enumerate(self.as_shapely()):
        for j, poly2 in enumerate(self.as_shapely()):
            if i == j:
                # polygon will obviously overlap with itself
                continue
            if poly1.intersects(poly2):
                self.set_logger_message(
                   f"{self.polygons[i].properties.get('name')} overlaps {self.polygons[j].properties.get('name')}.",
                   level="warning",
                )

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

from_file(file_path)

Read section polygon file.

Parameters:

Name Type Description Default
file_path str | Path

path to section polygon file.

 required
Source code in fm2prof\polygon_file.py 
500
501
502
503
504
505
506
507
508
def from_file(self, file_path: str | Path) -> None:
    """Read section polygon file.

    Args:
        file_path (str | Path): path to section polygon file.

    """
    super().from_file(file_path)
    self._validate_sections()

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_gridpoints_in_polygon(res_file, *, property_name='section', force_cache_invalidation=False)

Method to get faces and edges in section.

This method performs caching of the in-polygon classification results to avoid recalculating if the section file has not changed. The cache is invalidated if the section file is modified or if force_cache_invalidation is set to True.

Parameters:

Name Type Description Default
res_file str | Path

path to result (map) netcdf file.

 required
property_name Literal['region', 'section']

Property to use for classification. Defaults to section

 'section'
force_cache_invalidation bool

Force cache invalidation even if section file has not changed.

 False

Returns:

Type Description
GridPointsInPolygonResults

GridPointsInPolygonResults
Source code in fm2prof\polygon_file.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
def get_gridpoints_in_polygon(self,
    res_file: str | Path,
    *,
    property_name: Literal["region", "section"]="section",
    force_cache_invalidation:bool=False) -> GridPointsInPolygonResults:
    """Method to get faces and edges in section.

    This method performs caching of the in-polygon classification results
    to avoid recalculating if the section file has not changed. The cache
    is invalidated if the section file is modified or if `force_cache_invalidation`
    is set to `True`.

    Args:
        res_file (str | Path): path to result (map) netcdf file.
        property_name (Literal["region", "section"]): Property to use for classification. Defaults to section
        force_cache_invalidation (bool): Force cache invalidation even if section file has not changed.

    Returns:
        GridPointsInPolygonResults
    """
    return super().get_gridpoints_in_polygon(res_file,
                                          property_name=property_name,
                                          force_cache_invalidation=force_cache_invalidation)

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

get_points_in_polygon(points, property_name)

Method to determine in which polygon input points are.

Warning

This method is not applicable for large number of points. Only use for small number of points (e.g. cross-section locations).

Parameters:

Name Type Description Default
points ndarray

Array of shape (n_points, 2) containing x,y coordinates of points to classify.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

list[str]: List of polygon names in which the points are located. If a point is not located in any polygon, it is classified as 'undefined'.
Source code in fm2prof\polygon_file.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_points_in_polygon(self, points: np.ndarray, property_name: Literal["region", "section"]) -> list[str]:
    """Method to determine in which polygon input points are.

    Warning:
        This method is not applicable for large number of points.
        Only use for small number of points (e.g. cross-section locations).

    Args:
        points (np.ndarray): Array of shape (n_points, 2) containing x,y coordinates of points to classify.
        property_name (Literal['region', 'section']): Property to use for classification.

    Returns:
        list[str]: List of polygon names in which the points are located. If a point is not located
                   in any polygon, it is classified as 'undefined'.
    """
    # Convert to shapely point
    points = [shapely.Point(xy) for xy in points]
    points_regions = [self.undefined] * len(points)

    # Assign point to region
    for i, point in enumerate(points):
        for j, polygon in enumerate(self.as_shapely()):
            if point.within(polygon):
                points_regions[i] = self.polygons[j].properties.get(property_name)
                break

    return points_regions

meshkernel_inpolygon(res_file, dtype, property_name)

Get grid points in polygon.

Parameters:

Name Type Description Default
res_file str | Path

Path to result (map) netcdf file.

 required
dtype Literal['face', 'edge', 'node']

Type of grid points to retrieve.

 required
property_name Literal['region', 'section']

Property to use for classification.

 required

Returns:

Type Description
list[str]

pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
Source code in fm2prof\polygon_file.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def meshkernel_inpolygon(self,
    res_file: str | Path,  # path to result (map) netcdf file
    dtype: Literal["face", "edge", "node"],
    property_name: Literal["region", "section"],
    ) -> list[str]:
    """Get grid points in polygon.

    Args:
        res_file (str | Path): Path to result (map) netcdf file.
        dtype (Literal["face", "edge", "node"]): Type of grid points to retrieve.
        property_name (Literal["region", "section"]): Property to use for classification.

    Returns:
        pd.DataFrame | dict: DataFrame or dictionary containing grid points in polygon.
    """
    # Step 1
    # Construct Meshkernel grid
    mk = MeshKernel(projection=ProjectionType.CARTESIAN)
    mesh2d_input = mk.mesh2d_get()

    fmdata = FMDataImporter(res_file)
    mesh2d_input.node_x = fmdata.get_variable("mesh2d_node_x")
    mesh2d_input.node_y = fmdata.get_variable("mesh2d_node_y")
    mesh2d_input.edge_nodes = fmdata.get_variable("mesh2d_edge_nodes").flatten() - 1

    mk.mesh2d_set(mesh2d_input)

    # Step 2: perform point-in-polygon classification
    nodes_in_polygon: list[str] = [self.undefined] * len(mesh2d_input.node_x)  # default to undefined

    for i, mk_polygon in enumerate(self.as_meshkernel()):
        self.set_logger_message(
            f"  | Classifying {dtype}s in polygon {self.polygons[i].properties.get('name')} "
            f"({i+1}/{len(self.polygons)})",
            level="info")
        indices: list[float] = mk.mesh2d_get_nodes_in_polygons(mk_polygon, inside=True).tolist()
        for j in indices:
            nodes_in_polygon[j] = self.polygons[i].properties.get(property_name)

    # Step 3: map to faces or edges if needed
    if dtype == "node":
        output = nodes_in_polygon
    elif dtype == "face":
        face_map: np.ndarray = fmdata.get_variable("mesh2d_face_nodes").T[0] -1
        node_to_face: list[str] = [self.undefined] * len(face_map)
        for face_index, map_index in enumerate(face_map.tolist()):
            node_to_face[int(face_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_face
    elif dtype == "edge":
        # only internal edges!
        internal_edges = fmdata.get_variable("mesh2d_edge_type")[:] == 1
        edge_map: np.ndarray = fmdata.get_variable("mesh2d_edge_nodes").T[0] -1
        edge_map = edge_map[internal_edges]
        node_to_edge: list[str] = [self.undefined] * len(edge_map)
        for edge_index, map_index in enumerate(edge_map.tolist()):
            node_to_edge[int(edge_index)] = nodes_in_polygon[int(map_index)]
        output = node_to_edge

    return output

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

Base classes and data containers.

ElapsedFileFormatter()

Bases: ElapsedFormatter

Elapsed file formatter class.

Instantiate an ElapsedFileFormatter object.

Source code in fm2prof\common.py 
165
166
167
168
169
170
171
172
173
174
175
def __init__(self) -> None:
    """Instantiate an ElapsedFileFormatter object."""
    super().__init__()
    self._resetStyle = ""
    self._colors = {
        "INFO": ["", ""],
        "DEBUG": ["", ""],
        "WARNING": ["", ""],
        "ERROR": ["", ""],
        "RESET": "",
    }

pbar: None | tqdm.tqdm | tqdm.std.tqdm property writable

Progress bar.

finish_task()

Finish task.

Source code in fm2prof\common.py 
140
141
142
def finish_task(self) -> None:
    """Finish task."""
    self.__new_iteration = -1

format(record)

Format log record.

Source code in fm2prof\common.py 
86
87
88
89
90
91
92
93
94
def format(self, record: LogRecord) -> str:
    """Format log record."""
    if self._intro:
        return self.__format_intro(record)
    if self.__new_iteration > 0:
        return self.__format_header(record)
    if self.__new_iteration == -1:
        return self.__format_footer(record)
    return self.__format_message(record)

get_elapsed_time()

Get elapsed time in seconds.

Source code in fm2prof\common.py 
155
156
157
158
159
def get_elapsed_time(self) -> float:
    """Get elapsed time in seconds."""
    current_time = datetime.now()
    elapsed_time = current_time - self.start_time
    return elapsed_time.total_seconds()

new_task()

Reset ElapsedTimeFormatter.

Source code in fm2prof\common.py 
135
136
137
138
def new_task(self) -> None:
    """Reset ElapsedTimeFormatter."""
    self.__new_iteration = 1
    self.__reset()

set_intro(flag=True)

Indicate intro section for formatter.

Source code in fm2prof\common.py 
151
152
153
def set_intro(self, flag: bool = True) -> None:  # noqa: FBT001, FBT002
    """Indicate intro section for formatter."""
    self._intro = flag

set_number_of_iterations(n)

Set numbber of iterations.

Source code in fm2prof\common.py 
144
145
146
147
148
149
def set_number_of_iterations(self, n: int) -> None:
    """Set numbber of iterations."""
    if n < 1:
        err_msg = "Total number of iterations should be higher than zero"
        raise ValueError(err_msg)
    self.number_of_iterations = n

start_new_iteration(pbar=None)

Start a new iteration with a progress bar.

Source code in fm2prof\common.py 
129
130
131
132
133
def start_new_iteration(self, pbar: tqdm.tqdm | None = None) -> None:
    """Start a new iteration with a progress bar."""
    self.current_iteration += 1
    self.new_task()
    self.pbar = pbar

ElapsedFormatter()

ElapsedFormatter class.

Instantiate an ElapsedFormatter object.

Source code in fm2prof\common.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __init__(self) -> None:
    """Instantiate an ElapsedFormatter object."""
    self.start_time = time()
    self.number_of_iterations: int = 1
    self.current_iteration: int = 0
    self._pbar: tqdm.tqdm = None
    self._resetStyle = Style.RESET_ALL
    self._colors = {
        "INFO": [Back.BLUE, Fore.BLUE],
        "DEBUG": [Back.CYAN, Fore.CYAN],
        "WARNING": [Back.YELLOW + Fore.BLACK, Fore.YELLOW],
        "ERROR": [Back.RED, Fore.RED],
        "CRITICAL": [Back.GREEN, Fore.GREEN],
    }

    colorama.init()

    # saves amount of errors / warnings
    self._loglibrary: dict = {"ERROR": 0, "WARNING": 0}

pbar: None | tqdm.tqdm | tqdm.std.tqdm property writable

Progress bar.

finish_task()

Finish task.

Source code in fm2prof\common.py 
140
141
142
def finish_task(self) -> None:
    """Finish task."""
    self.__new_iteration = -1

format(record)

Format log record.

Source code in fm2prof\common.py 
86
87
88
89
90
91
92
93
94
def format(self, record: LogRecord) -> str:
    """Format log record."""
    if self._intro:
        return self.__format_intro(record)
    if self.__new_iteration > 0:
        return self.__format_header(record)
    if self.__new_iteration == -1:
        return self.__format_footer(record)
    return self.__format_message(record)

get_elapsed_time()

Get elapsed time in seconds.

Source code in fm2prof\common.py 
155
156
157
158
159
def get_elapsed_time(self) -> float:
    """Get elapsed time in seconds."""
    current_time = datetime.now()
    elapsed_time = current_time - self.start_time
    return elapsed_time.total_seconds()

new_task()

Reset ElapsedTimeFormatter.

Source code in fm2prof\common.py 
135
136
137
138
def new_task(self) -> None:
    """Reset ElapsedTimeFormatter."""
    self.__new_iteration = 1
    self.__reset()

set_intro(flag=True)

Indicate intro section for formatter.

Source code in fm2prof\common.py 
151
152
153
def set_intro(self, flag: bool = True) -> None:  # noqa: FBT001, FBT002
    """Indicate intro section for formatter."""
    self._intro = flag

set_number_of_iterations(n)

Set numbber of iterations.

Source code in fm2prof\common.py 
144
145
146
147
148
149
def set_number_of_iterations(self, n: int) -> None:
    """Set numbber of iterations."""
    if n < 1:
        err_msg = "Total number of iterations should be higher than zero"
        raise ValueError(err_msg)
    self.number_of_iterations = n

start_new_iteration(pbar=None)

Start a new iteration with a progress bar.

Source code in fm2prof\common.py 
129
130
131
132
133
def start_new_iteration(self, pbar: tqdm.tqdm | None = None) -> None:
    """Start a new iteration with a progress bar."""
    self.current_iteration += 1
    self.new_task()
    self.pbar = pbar

FM2ProfBase(logger=None, inifile=None)

Base class for FM2PROF types.

Implements methods for logging, project specific parameters

Instatiate a FM2ProfBase object.

logger (Logger | None, optional): Logger . Defaults to None.
inifile (IniFile | None, optional): IniFile instance. Defaults to None.
Source code in fm2prof\common.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
def __init__(self, logger: Logger | None = None, inifile: IniFile | None = None) -> None:
    """Instatiate a FM2ProfBase object.

    Args:
    ----
        logger (Logger | None, optional): Logger . Defaults to None.
        inifile (IniFile | None, optional): IniFile instance. Defaults to None.

    """
    if logger:
        self.set_logger(logger)
    if inifile:
        self.set_inifile(inifile)

create_logger() staticmethod

Create logger instance.

Source code in fm2prof\common.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@staticmethod
def create_logger() -> Logger:
    """Create logger instance."""
    # Create logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    # create formatter
    logger.__logformatter = ElapsedFormatter()  # noqa: SLF001
    logger._Filelogformatter = ElapsedFileFormatter()  # noqa: SLF001

    # create console handler
    if TqdmLoggingHandler not in map(type, logger.handlers):
        ch = TqdmLoggingHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(logger.__logformatter)  # noqa: SLF001
        logger.addHandler(ch)

    return logger

finish_log_task()

Use this method to finish task.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
297
298
299
300
301
302
303
304
def finish_log_task(self) -> None:
    """Use this method to finish task.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().finish_task()
    self.set_logger_message()
    self.pbar = None

get_filelogformatter()

Return file log formatter.

Source code in fm2prof\common.py 
310
311
312
def get_filelogformatter(self) -> ElapsedFormatter:
    """Return  file log formatter."""
    return self.get_logger()._Filelogformatter  # noqa: SLF001

get_inifile()

Get the inifile object.

Source code in fm2prof\common.py 
344
345
346
def get_inifile(self) -> IniFile:
    """Get the inifile object."""
    return self.__iniFile

get_logformatter()

Return log formatter.

Source code in fm2prof\common.py 
306
307
308
def get_logformatter(self) -> ElapsedFormatter:
    """Return log formatter."""
    return self.get_logger().__logformatter  # noqa: SLF001

get_logger()

Use this method to return logger object.

Source code in fm2prof\common.py 
226
227
228
def get_logger(self) -> Logger:
    """Use this method to return logger object."""
    return self.__logger

set_inifile(inifile=None)

Use this method to set configuration file object.

For loading from file, use load_inifile instead

inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.
Source code in fm2prof\common.py 
332
333
334
335
336
337
338
339
340
341
342
def set_inifile(self, inifile: IniFile = None) -> None:
    """Use this method to set configuration file object.

    For loading from file, use ``load_inifile`` instead

    Args:
    ----
        inifile (IniFile): inifile object. Obtain using e.g. ``get_inifile``.

    """
    self.__iniFile = inifile

set_logfile(output_dir, filename='fm2prof.log')

Set log file.

output_dir (str): _description_
filename (str, optional): _description_. Defaults to "fm2prof.log".
Source code in fm2prof\common.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def set_logfile(self, output_dir: str | Path, filename: str = "fm2prof.log") -> None:
    """Set log file.

    Args:
    ----
        output_dir (str): _description_
        filename (str, optional): _description_. Defaults to "fm2prof.log".

    """
    # create file handler
    if not output_dir:
        err_msg = "output_dir is required."
        raise ValueError(err_msg)
    fh = logging.FileHandler(Path(output_dir).joinpath(filename), encoding="utf-8")
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(self.get_logger()._Filelogformatter)  # noqa: SLF001
    self.__logger.addHandler(fh)

set_logger(logger)

Use to set logger.

logger (Logger): Logger instance
Source code in fm2prof\common.py 
230
231
232
233
234
235
236
237
238
239
240
241
def set_logger(self, logger: Logger) -> None:
    """Use to set logger.

    Args:
    ----
        logger (Logger): Logger instance

    """
    if not isinstance(logger, Logger):
        err_msg = "logger should be instance of Logger class"
        raise TypeError(err_msg)
    self.__logger = logger

set_logger_message(err_mssg='', level='info', *, header=False)

Set message to logger if this is set.

err_mssg (str, optional): Error message to log. Defaults to "".
level (str, optional): Log level. Defaults to "info".
header (bool, optional): Set error message as header. Defaults to False.
Source code in fm2prof\common.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def set_logger_message(
    self,
    err_mssg: str = "",
    level: str = "info",
    *,
    header: bool = False,
) -> None:
    """Set message to logger if this is set.

    Args:
    ----
        err_mssg (str, optional): Error message to log. Defaults to "".
        level (str, optional): Log level. Defaults to "info".
        header (bool, optional): Set error message as header. Defaults to False.

    """
    if not self.__logger:
        return

    if header:
        self.get_logformatter().set_intro(True)
        self.get_logger()._Filelogformatter.set_intro(True)  # noqa: SLF001
    else:
        self.get_logformatter().set_intro(False)
        self.get_logger()._Filelogformatter.set_intro(False)  # noqa: SLF001

    if level.lower() not in ["info", "debug", "warning", "error", "critical"]:
        err_msg = f"{level.lower()} is not valid logging level."
        raise ValueError(err_msg)

    if level.lower() == "info":
        self.__logger.info(err_mssg)
    elif level.lower() == "debug":
        self.__logger.debug(err_mssg)
    elif level.lower() == "warning":
        self.__logger.warning(err_mssg)
    elif level.lower() == "error":
        self.__logger.error(err_mssg)
    elif level.lower() in ["succes", "critical"]:
        self.__logger.critical(err_mssg)

start_new_log_task(task_name='NOT DEFINED', pbar=None)

Use this method to start a new task. Will reset the internal clock.

:param task_name: task name, will be displayed in log message

Source code in fm2prof\common.py 
284
285
286
287
288
289
290
291
292
293
294
295
def start_new_log_task(
    self,
    task_name: str = "NOT DEFINED",
    pbar: tqdm.tqdm = None,
) -> None:
    """Use this method to start a new task. Will reset the internal clock.

    :param task_name: task name, will be displayed in log message
    """
    self.get_logformatter().start_new_iteration(pbar=pbar)
    self.get_filelogformatter().start_new_iteration(pbar=pbar)
    self.set_logger_message(f"Starting new task: {task_name}")

FrictionTable(level, friction)

Container for friction table.

Instantiate a FrictionTable object.

Source code in fm2prof\common.py 
352
353
354
355
356
def __init__(self, level: np.ndarray, friction: np.ndarray) -> None:
    """Instantiate a FrictionTable object."""
    if self._validate_input(level, friction):
        self.level = level
        self.friction = friction

interpolate(new_z)

Interpolate friction.

Parameters:

Name Type Description Default
new_z ndarray

description

 required
Source code in fm2prof\common.py 
358
359
360
361
362
363
364
365
def interpolate(self, new_z: np.ndarray) -> None:
    """Interpolate friction.

    Args:
        new_z (np.ndarray): _description_
    """
    self.friction = np.interp(new_z, self.level, self.friction)
    self.level = new_z

TqdmLoggingHandler()

Bases: StreamHandler

Logging handler for tqdm package.

Instantiate a TqdmLoggingHandler.

Source code in fm2prof\common.py 
29
30
31
def __init__(self) -> None:
    """Instantiate a TqdmLoggingHandler."""
    super().__init__()

emit(record)

Write progressbar to logstream.

Source code in fm2prof\common.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def emit(self, record: LogRecord) -> None:
    """Write progressbar to logstream."""
    try:
        msg = self.format(record)
        if self.formatter.pbar:
            self.formatter.pbar.write(msg)
        else:
            stream = self.stream
            stream.write(msg + self.terminator)

        self.flush()
    except Exception:
        self.handleError(record)

Contains functions used for the emulation/reduction of 2D models to 1D models for Delft3D FM (D-Hydro).

classify_with_regions(cssdata, time_independent_data, edge_data, css_regions)

Assign cross-section id's based on region polygons.

Within a region, assignment will be done by k nearest neighbour

Source code in fm2prof\nearest_neighbour.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def classify_with_regions(
    cssdata: dict,
    time_independent_data: pd.DataFrame,
    edge_data: dict,
    css_regions: list[str],
    ) -> tuple[pd.DataFrame, dict]:
    """Assign cross-section id's based on region polygons.

    Within a region, assignment will be done by k nearest neighbour
    """
    time_independent_data["sclass"] = time_independent_data["region"].astype(str)

    # Nearest Neighbour within regions
    for region in time_independent_data["region"].unique():
        # Select cross-sections within this region
        css_xy = cssdata["xy"][np.array(css_regions) == region]
        css_id = cssdata["id"][np.array(css_regions) == region]

        if len(css_id) == 0:
            # no cross-sections in this region
            continue
        # Select 2d points within region
        node_mask = time_independent_data["region"] == region
        x_2d_node = time_independent_data["x"][node_mask]
        y_2d_node = time_independent_data["y"][node_mask]

        edge_mask = np.array(edge_data["region"]) == region
        x_2d_edge = edge_data["x"][edge_mask]
        y_2d_edge = edge_data["y"][edge_mask]

        # Do Nearest Neighour
        neigh = _get_class_tree(css_xy, css_id)
        css_2d_nodes = neigh.predict(np.array([x_2d_node, y_2d_node]).T)
        css_2d_edges = neigh.predict(np.array([x_2d_edge, y_2d_edge]).T)

        # Update data in main structures
        time_independent_data.loc[node_mask, "sclass"] = css_2d_nodes  # sclass = cross-section id

        edge_data["sclass"][edge_mask] = css_2d_edges

    return time_independent_data, edge_data

classify_without_regions(cssdata, time_independent_data, edge_data)

Classify without regions.

Source code in fm2prof\nearest_neighbour.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def classify_without_regions(
    cssdata: dict,
    time_independent_data: pd.DataFrame,
    edge_data: dict,
    ) -> tuple[pd.DataFrame, dict]:
    """Classify without regions."""
    # Create a class identifier to map points to cross-sections
    neigh = _get_class_tree(cssdata["xy"], cssdata["id"])

    # Expand time-independent dataset with cross-section names
    time_independent_data["sclass"] = neigh.predict(
        np.array([time_independent_data["x"], time_independent_data["y"]]).T,
    )

    # Assign cross-section names to edge coordinates as well
    edge_data["sclass"] = neigh.predict(np.array([edge_data["x"], edge_data["y"]]).T)

    return time_independent_data, edge_data

get_centre_values(location, x, y, waterdepth, waterlevel)

Find output point closest to x,y location, output depth and water level as nd arrays.

Source code in fm2prof\nearest_neighbour.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def get_centre_values(
    location: np.array,
    x: float,
    y: float,
    waterdepth: pd.DataFrame,
    waterlevel: pd.DataFrame,
    ) -> tuple[np.ndarray, np.ndarray]:
    """Find output point closest to x,y location, output depth and water level as nd arrays."""
    nn = NearestNeighbors(n_neighbors=1, algorithm="ball_tree").fit(np.array([x, y]).T)

    # conversion to 2d array, as 1d arrays are deprecated for kneighbors
    location_array = np.array(location).reshape(1, -1)
    (_, index) = nn.kneighbors(location_array)

    # retrieve cell characteristic waterdepth
    centre_depth = waterdepth.iloc[index[0]].to_numpy()
    centre_level = waterlevel.iloc[index[0]].to_numpy()

    # When starting from a dry bed, the centre_level may have nan values
    #
    bed_level = np.nanmin(centre_level - centre_depth)
    centre_level[np.isnan(centre_level)] = bed_level

    return centre_depth[0], centre_level[0]