src.digitaltwin.utils

This script provides utility functions for logging configuration and geospatial data manipulation.

Attributes

log

FuncArgsT

FuncKwargsT

FuncReturnT

Classes

LogLevel

Enum class representing different logging levels mapped to their corresponding numeric values from the

Functions

log_execution_info(→ None)

Log a debug message indicating the execution of the function in the script.

setup_logging(→ None)

Configure the root logger with the specified log level and formats, capture warnings, and exclude specific

get_catchment_area(→ geopandas.GeoDataFrame)

Convert the coordinate reference system (CRS) of the catchment area GeoDataFrame to the specified CRS.

get_nz_boundary(→ geopandas.GeoDataFrame)

Get the boundary of New Zealand in the specified Coordinate Reference System (CRS).

retry_function(→ FuncReturnT)

Retry a function a number of times if an exception is raised.

Module Contents

src.digitaltwin.utils.log
class src.digitaltwin.utils.LogLevel

Bases: enum.IntEnum

Enum class representing different logging levels mapped to their corresponding numeric values from the logging library.

CRITICAL

The critical logging level. Corresponds to logging.CRITICAL (50).

Type:

int

ERROR

The error logging level. Corresponds to logging.ERROR (40).

Type:

int

WARNING

The warning logging level. Corresponds to logging.WARNING (30).

Type:

int

INFO

The info logging level. Corresponds to logging.INFO (20).

Type:

int

DEBUG

The debug logging level. Corresponds to logging.DEBUG (10).

Type:

int

NOTSET

The not-set logging level. Corresponds to logging.NOTSET (0).

Type:

int

CRITICAL = 50
ERROR = 40
WARNING = 30
INFO = 20
DEBUG = 10
NOTSET = 0
src.digitaltwin.utils.log_execution_info() None

Log a debug message indicating the execution of the function in the script.

src.digitaltwin.utils.setup_logging(log_level: LogLevel = LogLevel.INFO) None

Configure the root logger with the specified log level and formats, capture warnings, and exclude specific loggers from propagating their messages to the root logger. Additionally, log a debug message indicating the execution of the function in the script.

Parameters:

log_level (LogLevel = LogLevel.DEBUG) – The log level to set for the root logger. Defaults to LogLevel.DEBUG. The available logging levels and their corresponding numeric values are: - LogLevel.CRITICAL (50) - LogLevel.ERROR (40) - LogLevel.WARNING (30) - LogLevel.INFO (20) - LogLevel.DEBUG (10) - LogLevel.NOTSET (0)

src.digitaltwin.utils.get_catchment_area(catchment_area: geopandas.GeoDataFrame, to_crs: int = 2193) geopandas.GeoDataFrame

Convert the coordinate reference system (CRS) of the catchment area GeoDataFrame to the specified CRS.

Parameters:
  • catchment_area (gpd.GeoDataFrame) – A GeoDataFrame representing the catchment area.

  • to_crs (int = 2193) – Coordinate Reference System (CRS) code to convert the catchment area to. Default is 2193.

Returns:

A GeoDataFrame representing the catchment area with the transformed CRS.

Return type:

gpd.GeoDataFrame

src.digitaltwin.utils.get_nz_boundary(engine: sqlalchemy.engine.Engine, to_crs: int = 2193) geopandas.GeoDataFrame

Get the boundary of New Zealand in the specified Coordinate Reference System (CRS).

Parameters:
  • engine (Engine) – The engine used to connect to the database.

  • to_crs (int = 2193) – Coordinate Reference System (CRS) code to which the boundary will be converted. Default is 2193.

Returns:

A GeoDataFrame representing the boundary of New Zealand in the specified CRS.

Return type:

gpd.GeoDataFrame

src.digitaltwin.utils.FuncArgsT
src.digitaltwin.utils.FuncKwargsT
src.digitaltwin.utils.FuncReturnT
src.digitaltwin.utils.retry_function(func: Callable[[FuncArgsT, FuncKwargsT], FuncReturnT], max_retries: int, base_retry_delay: float, expected_exceptions: Type[Exception] | Tuple[Type[Exception]], *args: FuncArgsT, **kwargs: FuncKwargsT) FuncReturnT

Retry a function a number of times if an exception is raised.

Parameters:
  • func (Callable[[FuncArgsT, FuncKwargsT], FuncReturnT]) – The function to call and retry if it fails. Takes *args and **kwargs as arguments.

  • max_retries (int) – The maximum number of times to retry the function before allowing the exception to propagate

  • base_retry_delay (float) – The delay in seconds between retries. Each subsequent retry becomes extended by this amount.

  • expected_exceptions (Type[Exception] | Tuple[Type[Exception]]) – The exceptions that are expected to be thrown, and so will be caught. Any others will be propagated.

  • *args (FuncArgsT) – The standard arguments for func.

  • **kwargs (FuncKwargsT) – The keyword arguments for func.

Returns:

The result of func(*args, **kwargs).

Return type:

FuncReturnT