improver.clustering.cluster_sources_utils module

Utilities for parsing and querying cluster_sources cube attributes.

find_nearest_forecast_period_gte(forecast_periods, target_fp)

Find the nearest forecast period >= target_fp from a set of forecast periods.

This is used for transition handling when querying secondary mappings: if an exact forecast period is not available, we find the next earliest one.

Parameters:

• forecast_periods (Optional[set[int]]) – Set of available forecast periods (in seconds). If None or empty, returns (target_fp, False).

• target_fp (int) – The target forecast period (in seconds) to find.

Returns:

• nearest_fp: The smallest forecast period from forecast_periods that is >= target_fp, or target_fp if none exist or forecast_periods is empty.

• use_secondary: Boolean indicating whether the nearest_fp is from forecast_periods (True) or equals target_fp (False).

Return type:

Tuple of (nearest_fp, use_secondary)

get_source_for_forecast_period(cluster_sources, realization_idx, fp_seconds)

Determine which forecast source (model) is active for a given realization and forecast period.

Uses transition handling: if the exact forecast period is not found in cluster_sources for this realization, we look for the largest period that is still <= the target period (conservative: stay with last-known source).

Parameters:

• cluster_sources (dict) – Dictionary mapping cluster/realization indices (as strings or ints) to model sources. Format: {realization_idx: {model_name: [fp1, fp2, …]}}

• realization_idx (int) – The realization/cluster index to query.

• fp_seconds (int) – The forecast period in seconds.

Return type:

Optional[str]

Returns:

The name of the forecast source (model) active at this forecast period, or None if the realization is not found or no source covers this period.

Example

>>> sources = {
...     "0": {
...         "uk_ens": [3600, 21600, 43200, 86400, 129600, 172800],
...         "gl_ens": [432000, 475200, 518400],
...     }
... }
>>> get_source_for_forecast_period(sources, "0", 90000)
'uk_ens'
>>> get_source_for_forecast_period(sources, "0", 432000)
'gl_ens'
>>> get_source_for_forecast_period(
...     sources, "0", 400000
... )  # Between uk_ens end & gl_ens start
'uk_ens'  # Returns last-known source

parse_cluster_sources_attribute(cube)

Parse the cluster_sources dictionary from a cube attribute.

The cluster_sources attribute is typically set by RealizationClusterAndMatch and consumed by RealizationSelection, StochasticNoise, and temporal interpolation plugins to determine which forecast source (model) was used for each cluster at each forecast period.

Parameters:

cube – A cube containing the cluster_sources attribute as a JSON string.

Return type:

dict

Returns:

Dictionary mapping cluster indices (as strings) to model mappings. Format: {cluster_idx: {model_name: [forecast_periods_in_seconds]}}. If the cluster_sources attribute is not present, returns an empty dictionary.

Raises:

• ValueError – If the cluster_sources attribute is malformed JSON.

• ValueError – If the attribute is not a dictionary or JSON string.

• ValueError – If the parsed JSON is not a dictionary.