Financial Helper Functions

Contains functions for financial computations.

calculate_activity_npv

calculate_activity_npv(activity_id, activity_area)

Determines the total NPV of an activity by calculating the individual NPV of the NCS pathways that constitute the activity.

The NPV per hectare for an NCS pathway is determined based on the value specified in the NPV PWL Manager.

Parameters:

Name	Type	Description	Default
activity_id	str	The ID of the specific activity. The function will check whether the NPV rate had been defined for pathways that constitute the activity.	required
activity_area	float	The area of the activity in hectares.	required

Returns:

Type	Description
float	Returns the total NPV of the activity, or -1.0 if the activity does not exist or if found, lacks pathways or if the NPV rate for all pathways has not been specified.

Source code in src/cplus_plugin/lib/financials.py

def calculate_activity_npv(activity_id: str, activity_area: float) -> float:
    """Determines the total NPV of an activity by calculating
    the individual NPV of the NCS pathways that constitute
    the activity.

    The NPV per hectare for an NCS pathway is determined based
    on the value specified in the NPV PWL Manager.

    :param activity_id: The ID of the specific activity. The
    function will check whether the NPV rate had been defined
    for pathways that constitute the activity.
    :type activity_id: str

    :param activity_area: The area of the activity in hectares.
    :type activity_area: float

    :returns: Returns the total NPV of the activity, or -1.0
    if the activity does not exist or if found, lacks pathways
    or if the NPV rate for all pathways has not been specified.
    :rtype: float
    """
    activity = settings_manager.get_activity(activity_id)
    if activity is None or len(activity.pathways) == 0:
        return -1.0

    npv_collection = settings_manager.get_npv_collection()
    if npv_collection is None:
        return -1.0

    pathway_npv_values = []
    for pathway in activity.pathways:
        pathway_npv = npv_collection.pathway_npv(str(pathway.uuid))
        if pathway_npv is None:
            continue

        if pathway_npv.params is None or pathway_npv.params.absolute_npv is None:
            continue

        pathway_npv_values.append(pathway_npv.params.absolute_npv)

    if len(pathway_npv_values) == 0:
        return -1.0

    return float(sum(pathway_npv_values) * activity_area)

compute_discount_value

compute_discount_value(revenue, cost, year, discount)

Calculates the discounted value for the given year.

Parameters:

Name	Type	Description	Default
revenue	float	Projected total revenue.	required
cost	float	Projected total costs.	required
year	int	Relative year i.e. between 1 and 99.	required
discount	float	Discount value as a percent i.e. between 0 and 100.	required

Returns:

Type	Description
float	The discounted value for the given year.

Source code in src/cplus_plugin/lib/financials.py

def compute_discount_value(
    revenue: float, cost: float, year: int, discount: float
) -> float:
    """Calculates the discounted value for the given year.

    :param revenue: Projected total revenue.
    :type revenue: float

    :param cost: Projected total costs.
    :type cost: float

    :param year: Relative year i.e. between 1 and 99.
    :type year: int

    :param discount: Discount value as a percent i.e. between 0 and 100.
    :type discount: float

    :returns: The discounted value for the given year.
    :rtype: float
    """
    return (revenue - cost) / ((1 + discount / 100.0) ** (year - 1))

create_npv_pwls

create_npv_pwls(npv_collection, context, multi_step_feedback, feedback, target_crs_id, target_pixel_size, target_extent, on_finish_func=None, on_removed_func=None)

Creates constant raster layers based on the normalized NPV values for the specified NCS pathways.

Parameters:

Name	Type	Description	Default
npv_collection	NcsPathwayNpvCollection	The NCS pathway NPV collection containing the NPV parameters for NCS pathway.	required
context	QgsProcessingContext	Context information for performing the processing.	required
multi_step_feedback	QgsProcessingMultiStepFeedback	Feedback for updating the status of processing.	required
feedback	QgsProcessingFeedback	Underlying feedback object for communicating to the caller.	required
target_crs_id	str	CRS identifier of the target layers.	required
target_pixel_size	float	Pixel size of the target layer:	required
target_extent	str	Extent of the output layer as xmin, xmax, ymin, ymax.	required
on_finish_func	Callable	Function to be executed when a constant raster has been created.	None
on_removed_func	Callable	Function to be executed when a disabled NPV PWL has been removed.	None

Returns:

Type	Description
list	A list containing the processing results (as a dictionary) for each successful run.

Source code in src/cplus_plugin/lib/financials.py

def create_npv_pwls(
    npv_collection: NcsPathwayNpvCollection,
    context: QgsProcessingContext,
    multi_step_feedback: QgsProcessingMultiStepFeedback,
    feedback: QgsProcessingFeedback,
    target_crs_id: str,
    target_pixel_size: float,
    target_extent: str,
    on_finish_func: typing.Callable = None,
    on_removed_func: typing.Callable = None,
) -> typing.List:
    """Creates constant raster layers based on the normalized NPV values for
    the specified NCS pathways.

    :param npv_collection: The NCS pathway NPV collection containing the NPV
    parameters for NCS pathway.
    :type npv_collection: NcsPathwayNpvCollection

    :param context: Context information for performing the processing.
    :type context: QgsProcessingContext

    :param multi_step_feedback: Feedback for updating the status of processing.
    :type multi_step_feedback: QgsProcessingMultiStepFeedback

    :param feedback: Underlying feedback object for communicating to the caller.
    :type feedback: QgsProcessingFeedback

    :param target_crs_id: CRS identifier of the target layers.
    :type target_crs_id: str

    :param target_pixel_size: Pixel size of the target layer:
    :type target_pixel_size: float

    :param target_extent: Extent of the output layer as xmin, xmax, ymin, ymax.
    :type target_extent: str

    :param on_finish_func: Function to be executed when a constant raster
    has been created.
    :type on_finish_func: Callable

    :param on_removed_func: Function to be executed when a disabled NPV PWL has
    been removed.
    :type on_finish_func: Callable

    :returns: A list containing the processing results (as a dictionary) for
    each successful run.
    :rtype: list
    """
    base_dir = settings_manager.get_value(Settings.BASE_DIR)
    if not base_dir:
        log(message=tr("No base directory for saving NPV PWLs."), info=False)
        return []

    # Create NPV PWL subdirectory
    FileUtils.create_npv_pwls_dir(base_dir)

    # NPV PWL root directory
    npv_base_dir = f"{base_dir}/{PRIORITY_LAYERS_SEGMENT}/{NPV_PRIORITY_LAYERS_SEGMENT}"

    current_step = 0
    multi_step_feedback.setCurrentStep(current_step)

    results = []

    for i, pathway_npv in enumerate(npv_collection.mappings):
        if feedback.isCanceled():
            break

        if pathway_npv.pathway is None or pathway_npv.params is None:
            log(
                tr(
                    "Could not create or update NCS pathway NPV as NCS "
                    "pathway and NPV parameter information is missing."
                ),
                info=False,
            )

            current_step += 1
            multi_step_feedback.setCurrentStep(current_step)

            continue

        base_layer_name = clean_filename(
            pathway_npv.base_name.replace(" ", "_").lower()
        )

        # Delete existing NPV PWLs. Relevant layers will be re-created
        # where applicable.
        for del_npv_path in pathlib.Path(npv_base_dir).glob(f"*{base_layer_name}*"):
            try:
                log(f"{tr('Deleting')} - NPV PWL {del_npv_path}")
                pathlib.Path(del_npv_path).unlink()
            except OSError as os_ex:
                base_msg_tr = tr("Unable to delete NPV PWL")
                conclusion_msg_tr = tr(
                    "File will be deleted in subsequent processes if not locked"
                )
                log(
                    f"{base_msg_tr}: {os_ex.strerror}. {conclusion_msg_tr}.", info=False
                )

        # Delete if PWL previously existed and is now disabled
        if not pathway_npv.enabled:
            if npv_collection.remove_existing:
                # Delete corresponding PWL entry in the settings
                del_pwl = settings_manager.find_layer_by_name(pathway_npv.base_name)
                if del_pwl is not None:
                    pwl_id = del_pwl.get("uuid", None)
                    if pwl_id is not None:
                        settings_manager.delete_priority_layer(pwl_id)
                        if on_removed_func is not None:
                            on_removed_func(str(del_pwl["uuid"]))

                current_step += 1
                multi_step_feedback.setCurrentStep(current_step)
                continue

        # Output layer name
        npv_pwl_path = f"{npv_base_dir}/{base_layer_name}_{datetime.datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.tif"

        output_post_processing_func = None
        if on_finish_func is not None:
            output_post_processing_func = partial(
                on_finish_func, pathway_npv, npv_pwl_path
            )

        try:
            alg_params = {
                "EXTENT": target_extent,
                "TARGET_CRS": target_crs_id,
                "PIXEL_SIZE": target_pixel_size,
                "NUMBER": pathway_npv.params.normalized_npv,
                "OUTPUT": npv_pwl_path,
            }
            res = processing.run(
                "native:createconstantrasterlayer",
                alg_params,
                context=context,
                feedback=multi_step_feedback,
                onFinish=output_post_processing_func,
            )
            results.append(res)
        except QgsProcessingException as ex:
            err_tr = tr("Error creating NPV PWL")
            log(f"{err_tr} {npv_pwl_path}")

        current_step += 1
        multi_step_feedback.setCurrentStep(current_step)

    return results

---

Last update: June 9, 2025