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