Loss Function

Overview

The optimisation loss function is customisable and user-defined. The user must provide a loss function that accepts two positional input dictionaries:

Simulation data — containing the mesh cell centroids and all hydrodynamic results.
Ground-truth data — containing the observed values at fixed XY coordinates.

This design allows the use of any tabular format for the ground-truth data. Any resolved hydrodynamic variable (e.g., water depth, flow velocities, Reynolds stresses, etc.) can be used individually or combined to compose the loss value. Utility tools such as utils/find_closest_cell assist the user in matching observed points to simulated cells.

Return Values

The user-defined loss function must return:

Required: A single floating-point value representing the global loss.
Optional: A second output containing a list of point-wise errors.

The optional point-wise errors do not affect the calibration process and are stored in CSV files solely for visualisation purposes.

Although not directly implemented, users may define variable-specific error thresholds within the loss function that, once satisfied, return a null loss value and terminate the code execution.

Example:

if water_depth_error < 0.1 and velocity_error < 0.1:
    return 0.0, pointwise_errors

Or implement weighted loss components that prioritise certain variables:

loss = 0.7 * water_depth_error + 0.3 * velocity_error
return loss, pointwise_errors

Example Loss Function implementation:

def loss_function(
    simulation_data: dict,
    ground_truth_data: dict,
) -> tuple[float, pd.DataFrame]:
    """
    Extracts simulation data at calibration points and computes the RMSE for water depth.

    Args:
        simulation_data (dict): Contains simulation data (e.g., centroids and hydrodynamic properties).
        ground_truth_data (dict): Contains calibration points with known water depths.

    Returns:
        float: The sum of RMSE values for all calibration cross-sections.
        (pd.Series): Optional, a series of errors for each calibration point, which can be used for further analysis or visualization.
    """
    # Initialize an array to store RMSE values
    rmse_depth = []
    rmse_ux = []
    rmse_uy = []
    errors = pd.DataFrame(columns=['x', 'y', 'error_h', 'error_ux', 'error_uy'])

    for ground_truth in ground_truth_data:
        # Create an empty array for simulated cross-section data
        gt_numpy = ground_truth.to_numpy()
        sim_cs = {'cx':[], 'cy':[], 'h':[], 'ux':[], 'uy':[]}

        for i, (val_x, val_y, val_h) in enumerate(gt_numpy):
            # Find the closest simulation cells to the calibration point
            closest_cell_ids, cell_distance = utils.find_closest_cell(
                simulation_data['centroids'],
                (val_x, val_y),
                npoints=6,
                distance=True
            )

            # Interpolate flow variables using IDW (Inverse Distance Weighting)
            sim_h = simulation_data['hyd']['h'][closest_cell_ids]
            sim_ux = simulation_data['hyd']['ux'][closest_cell_ids]
            sim_uy = simulation_data['hyd']['uy'][closest_cell_ids]

            sim_cs['h'].append(np.average(sim_h, weights=1 / (cell_distance**2)))
            sim_cs['ux'].append(np.average(sim_ux, weights=1 / (cell_distance**2)))
            sim_cs['uy'].append(np.average(sim_uy, weights=1 / (cell_distance**2)))

            sim_cs['cx'].append(val_x)
            sim_cs['cy'].append(val_y)

        # Checks for negative water depth
        if any(h < 0 for h in sim_cs['h']):
            return 1e6, errors

        # Compute RMSE for this cross-section
        error_h = ground_truth['H [m]'] - sim_cs['h']
        error_ux = ground_truth['Ux [m/s]'] - sim_cs['ux']
        error_uy = ground_truth['Uy [m/s]'] - sim_cs['uy']

        # Store pointwise errors for potential further analysis or visualization
        new_rows = pd.DataFrame({
            'x': sim_cs['cx'],
            'y': sim_cs['cy'],
            'error_h': error_h,
            'error_ux': error_ux,
            'error_uy': error_uy
        })
        errors = pd.concat([errors, new_rows], ignore_index=True)

        # Compute RMSE values
        rmse_depth_i = np.sqrt(np.mean((ground_truth['H [m]'] - sim_cs['h'])**2))
        rmse_ux_i = np.sqrt(np.mean((ground_truth['Ux [m/s]'] - sim_cs['ux'])**2))
        rmse_uy_i = np.sqrt(np.mean((ground_truth['Uy [m/s]'] - sim_cs['uy'])**2))

        # Store RMSE values
        rmse_depth.append(rmse_depth_i)
        rmse_ux.append(rmse_ux_i)
        rmse_uy.append(rmse_uy_i)

    # Combine the RMSE values for each file
    mean_rmse_depth = np.mean(rmse_depth)
    mean_rmse_ux = np.mean(rmse_ux)
    mean_rmse_uy = np.mean(rmse_uy)

    loss_value = 0.5 * mean_rmse_depth + 0.25 * mean_rmse_ux + 0.25 * mean_rmse_uy

    ## It is possible to create a more complex loss function that combines multiple metrics (e.g., water depth and velocity errors) with different weights. The example above uses a simple weighted sum of RMSE values for water depth and velocity components.

    ## But it is also possible to return a simple loss value based on a single metric, such as water depth RMSE, if that is the primary focus of the calibration. In that case, the function would return only the RMSE for water depth:
    # loss_value = mean_rmse_depth

    ## The second return value is optional and can be used to save the error comparison points for all tried vectors, if specified in the user options.
    ## If the second value is not needed, simply return:
    # return loss_value, None

    return loss_value, errors

Input Dictionary Structure

`simulation_data`

Variable	Description	In `simulation_data`, accessible with:
water_depth	Array of simulated water depths per cell	`simulation_data['hyd']['h']`
velocity_x	Array of X-component flow velocities per cell	`simulation_data['hyd']['ux']`
velocity_y	Array of Y-component flow velocities per cell	`simulation_data['hyd']['uy']`
specific_discharge_x	Array of X-component specific discharge per cell	`simulation_data['hyd']['qx']`
specific_discharge_y	Array of Y-component specific discharge per cell	`simulation_data['hyd']['qy']`
turbulent_k	Array of turbulent kinetic energy per cell	`simulation_data['turb_k']`
turbulent_reynolds_xx	Array of turbulent Reynolds stresses in X direction	`simulation_data['turb_reynolds'][:, 0]`
turbulent_reynolds_yy	Array of turbulent Reynolds stresses in Z direction	`simulation_data['turb_reynolds'][:, 1]`

`ground_truth_data`

The ground-truth dictionary can hold any tabular structure with a header. Note that the coordinates must be retrievable as ground_truth_data['observations'][i]['x'] and ground_truth_data['observations'][i]['y'] for each observation point i. The observed variable values can be accessed similarly, e.g., ground_truth_data['observations'][i]['water_depth'].

Utility: `utils/find_closest_cell`

The find_closest_cell utility maps an observed point (X, Y) to the index of the nearest simulated mesh cell, based on the cell centroid coordinates stored in simulation_data.

cell_index = find_closest_cell(simulation_data, x, y)

This allows the user to extract the simulated hydrodynamic variables at the location closest to each observation point.

Notes

The loss function file must be placed at user_defined_configs/function_loss.py.
The function name must be loss_function.
Only the first return value (global loss) influences the calibration algorithm.
The optional second return value (point-wise errors) is exported as a CSV file for post-processing and visualisation only.
Users may implement early stopping by returning 0.0 as the loss value when predefined error thresholds are met for all variables of interest.
Further information can be found at Function Loss.