Skip to content

MESQUAL Folium KPICollection Visualization System

KPICollectionMapVisualizer

High-level KPI collection map visualizer for energy system analysis.

Main orchestrator for converting KPI collections into organized folium map visualizations. Handles KPI grouping, feature group creation, tooltip enhancement, and progress tracking. Designed to replicate and extend the functionality of the original KPIToMapVisualizerBase.

Supports multiple generators for complex visualizations (e.g., areas with text overlays, lines with arrow indicators) and provides sophisticated KPI organization and related KPI discovery for enhanced user experience.

Parameters:

Name Type Description Default
generators FoliumObjectGenerator | List[FoliumObjectGenerator]

Single generator or list of generators for visualization

required
study_manager StudyManager

StudyManager for enhanced KPI relationships and tooltips

None
include_related_kpis_in_tooltip bool

Add related KPIs to tooltip display

False
kpi_grouping_manager KPIGroupingManager

Custom grouping manager (optional)

None
**kwargs

Additional arguments passed to data items

{}

Examples:

Basic area visualization:

>>> visualizer = KPICollectionMapVisualizer(
...     generators=[
...         AreaGenerator(
...             AreaFeatureResolver(
...                 fill_color=PropertyMapper.from_kpi_value(color_scale),
...                 tooltip=True
...             )
...         )
...     ],
...     study_manager=study
... )
>>> feature_groups = visualizer.get_feature_groups(price_kpis)
>>> for fg in feature_groups:
...     fg.add_to(map)

Complex multi-layer visualization:

>>> visualizer = KPICollectionMapVisualizer(
...     generators=[
...         AreaGenerator(AreaFeatureResolver(...)),
...         TextOverlayGenerator(TextOverlayFeatureResolver(...))
...     ],
...     include_related_kpis_in_tooltip=True,
...     study_manager=study
... )
>>> visualizer.generate_and_add_feature_groups_to_map(
...     kpi_collection, folium_map, show='first'
... )

Flow visualization with arrows:

>>> visualizer = KPICollectionMapVisualizer(
...     generators=[
...         ArrowIconGenerator(ArrowIconFeatureResolver(...)),
...         TextOverlayGenerator(TextOverlayFeatureResolver(...))
...     ]
... )
Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
class KPICollectionMapVisualizer:
    """
    High-level KPI collection map visualizer for energy system analysis.

    Main orchestrator for converting KPI collections into organized folium map
    visualizations. Handles KPI grouping, feature group creation, tooltip
    enhancement, and progress tracking. Designed to replicate and extend
    the functionality of the original KPIToMapVisualizerBase.

    Supports multiple generators for complex visualizations (e.g., areas with
    text overlays, lines with arrow indicators) and provides sophisticated
    KPI organization and related KPI discovery for enhanced user experience.

    Args:
        generators: Single generator or list of generators for visualization
        study_manager: StudyManager for enhanced KPI relationships and tooltips
        include_related_kpis_in_tooltip: Add related KPIs to tooltip display
        kpi_grouping_manager: Custom grouping manager (optional)
        **kwargs: Additional arguments passed to data items

    Examples:
        Basic area visualization:
        >>> visualizer = KPICollectionMapVisualizer(
        ...     generators=[
        ...         AreaGenerator(
        ...             AreaFeatureResolver(
        ...                 fill_color=PropertyMapper.from_kpi_value(color_scale),
        ...                 tooltip=True
        ...             )
        ...         )
        ...     ],
        ...     study_manager=study
        ... )
        >>> feature_groups = visualizer.get_feature_groups(price_kpis)
        >>> for fg in feature_groups:
        ...     fg.add_to(map)

        Complex multi-layer visualization:
        >>> visualizer = KPICollectionMapVisualizer(
        ...     generators=[
        ...         AreaGenerator(AreaFeatureResolver(...)),
        ...         TextOverlayGenerator(TextOverlayFeatureResolver(...))
        ...     ],
        ...     include_related_kpis_in_tooltip=True,
        ...     study_manager=study
        ... )
        >>> visualizer.generate_and_add_feature_groups_to_map(
        ...     kpi_collection, folium_map, show='first'
        ... )

        Flow visualization with arrows:
        >>> visualizer = KPICollectionMapVisualizer(
        ...     generators=[
        ...         ArrowIconGenerator(ArrowIconFeatureResolver(...)),
        ...         TextOverlayGenerator(TextOverlayFeatureResolver(...))
        ...     ]
        ... )
    """

    def __init__(
            self,
            generators: FoliumObjectGenerator | List[FoliumObjectGenerator],
            study_manager: 'StudyManager' = None,
            include_related_kpis_in_tooltip: bool = False,
            kpi_grouping_manager: KPIGroupingManager = None,
            **kwargs
    ):
        self.generators: List[FoliumObjectGenerator] = generators if isinstance(generators, list) else [generators]
        self.study_manager = study_manager
        self.include_related_kpis_in_tooltip = include_related_kpis_in_tooltip

        self.grouping_manager = kpi_grouping_manager or KPIGroupingManager()

        # Enhanced tooltip if needed
        if self.include_related_kpis_in_tooltip:
            for g in self.generators:
                g.tooltip_generator = self._create_enhanced_tooltip_generator()
        self.kwargs = kwargs

    def generate_and_add_feature_groups_to_map(
            self,
            kpi_collection: KPICollection,
            folium_map: folium.Map,
            show: SHOW_OPTIONS = 'none',
            overlay: bool = False,
    ) -> list[folium.FeatureGroup]:
        fgs = self.get_feature_groups(kpi_collection, show=show, overlay=overlay)
        for fg in fgs:
            folium_map.add_child(fg)
        return fgs

    def get_feature_groups(
            self,
            kpi_collection: KPICollection,
            show: SHOW_OPTIONS = 'none',
            overlay: bool = False
    ) -> list[folium.FeatureGroup]:
        """
        Create feature groups for KPI collection with organized grouping.

        Main method that processes KPI collection through grouping, creates
        folium FeatureGroups, and applies all configured generators to create
        a complete map visualization.

        Args:
            kpi_collection: Collection of KPIs to visualize
            show: Which feature groups to show initially ('first', 'last', 'none')
            overlay: Whether feature groups should be overlay controls

        Returns:
            List of folium FeatureGroup objects ready to add to map
        """
        """Create feature groups for KPI collection, replicating original functionality."""
        from tqdm import tqdm
        from mesqual.utils.logging import get_logger

        logger = get_logger(__name__)
        feature_groups = []

        pbar = tqdm(kpi_collection, total=kpi_collection.size, desc=f'{self.__class__.__name__}')
        with pbar:
            kpi_groups = self.grouping_manager.get_kpi_groups(kpi_collection)
            for kpi_group in kpi_groups:
                group_name = self.grouping_manager.get_feature_group_name(kpi_group)

                if show == 'first':
                    show_fg = kpi_group == kpi_groups[0]
                elif show == 'last':
                    show_fg = kpi_group == kpi_groups[-1]
                else:
                    show_fg = False

                fg = folium.FeatureGroup(name=group_name, overlay=overlay, show=show_fg)

                for kpi in kpi_group:
                    data_item = KPIDataItem(kpi, kpi_collection, study_manager=self.study_manager, **self.kwargs)
                    for generator in self.generators:
                        if self.include_related_kpis_in_tooltip:
                            _tmp = generator.feature_resolver.property_mappers.get('tooltip', None)
                            generator.feature_resolver.property_mappers['tooltip'] = self._create_enhanced_tooltip_generator()
                        try:
                            generator.generate(data_item, fg)
                        except Exception as e:
                            logger.warning(
                                f'Exception while trying to add KPI {kpi.name} to FeatureGroup {group_name}: {e}'
                            )
                        finally:
                            if self.include_related_kpis_in_tooltip:
                                if _tmp is not None:
                                    generator.feature_resolver.property_mappers['tooltip'] = _tmp
                                else:
                                    generator.feature_resolver.property_mappers.pop('tooltip')
                    pbar.update(1)

                feature_groups.append(fg)

        return feature_groups

    def _create_enhanced_tooltip_generator(self) -> PropertyMapper:
        """Create tooltip generator that includes related KPIs."""

        def generate_tooltip(data_item: KPIDataItem) -> str:

            kpi = data_item.kpi
            kpi_name = kpi.get_kpi_name_with_dataset_name()

            from mesqual.units import Units
            kpi_quantity = Units.get_quantity_in_pretty_unit(kpi.quantity)
            kpi_text = Units.get_pretty_text_for_quantity(kpi_quantity, thousands_separator=' ')

            html = '<table style="border-collapse: collapse;">\n'
            html += f'  <tr><td style="padding: 4px 8px;"><strong>{kpi_name}</strong></td>' \
                    f'<td style="text-align: right; padding: 4px 8px;">{kpi_text}</td></tr>\n'

            if self.include_related_kpis_in_tooltip and self.study_manager:
                related_groups = self.grouping_manager.get_related_kpi_groups(
                    kpi, self.study_manager
                )

                if any(not g.empty for g in related_groups.values()):
                    for name, group in related_groups.items():
                        if group.empty:
                            continue
                        html += "<tr><p>&nbsp;</p></tr>"
                        html += f'  <tr><th colspan="2" style="text-align: left; padding: 8px;">{name}</th></tr>\n'
                        for related_kpi in group:
                            related_kpi_name = related_kpi.get_kpi_name_with_dataset_name()
                            related_kpi_quantity = Units.get_quantity_in_pretty_unit(related_kpi.quantity)
                            related_kpi_value_text = Units.get_pretty_text_for_quantity(
                                related_kpi_quantity,
                                thousands_separator=' ',
                            )
                            html += f'  <tr><td style="padding: 4px 8px;">{related_kpi_name}</td>' \
                                    f'<td style="text-align: right; padding: 4px 8px;">{related_kpi_value_text}</td></tr>\n'

            html += '<br><p>&nbsp;</p></table>'
            return html

        return PropertyMapper(generate_tooltip)

get_feature_groups

get_feature_groups(kpi_collection: KPICollection, show: SHOW_OPTIONS = 'none', overlay: bool = False) -> list[FeatureGroup]

Create feature groups for KPI collection with organized grouping.

Main method that processes KPI collection through grouping, creates folium FeatureGroups, and applies all configured generators to create a complete map visualization.

Parameters:

Name Type Description Default
kpi_collection KPICollection

Collection of KPIs to visualize

required
show SHOW_OPTIONS

Which feature groups to show initially ('first', 'last', 'none')

'none'
overlay bool

Whether feature groups should be overlay controls

False

Returns:

Type Description
list[FeatureGroup]

List of folium FeatureGroup objects ready to add to map

Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def get_feature_groups(
        self,
        kpi_collection: KPICollection,
        show: SHOW_OPTIONS = 'none',
        overlay: bool = False
) -> list[folium.FeatureGroup]:
    """
    Create feature groups for KPI collection with organized grouping.

    Main method that processes KPI collection through grouping, creates
    folium FeatureGroups, and applies all configured generators to create
    a complete map visualization.

    Args:
        kpi_collection: Collection of KPIs to visualize
        show: Which feature groups to show initially ('first', 'last', 'none')
        overlay: Whether feature groups should be overlay controls

    Returns:
        List of folium FeatureGroup objects ready to add to map
    """
    """Create feature groups for KPI collection, replicating original functionality."""
    from tqdm import tqdm
    from mesqual.utils.logging import get_logger

    logger = get_logger(__name__)
    feature_groups = []

    pbar = tqdm(kpi_collection, total=kpi_collection.size, desc=f'{self.__class__.__name__}')
    with pbar:
        kpi_groups = self.grouping_manager.get_kpi_groups(kpi_collection)
        for kpi_group in kpi_groups:
            group_name = self.grouping_manager.get_feature_group_name(kpi_group)

            if show == 'first':
                show_fg = kpi_group == kpi_groups[0]
            elif show == 'last':
                show_fg = kpi_group == kpi_groups[-1]
            else:
                show_fg = False

            fg = folium.FeatureGroup(name=group_name, overlay=overlay, show=show_fg)

            for kpi in kpi_group:
                data_item = KPIDataItem(kpi, kpi_collection, study_manager=self.study_manager, **self.kwargs)
                for generator in self.generators:
                    if self.include_related_kpis_in_tooltip:
                        _tmp = generator.feature_resolver.property_mappers.get('tooltip', None)
                        generator.feature_resolver.property_mappers['tooltip'] = self._create_enhanced_tooltip_generator()
                    try:
                        generator.generate(data_item, fg)
                    except Exception as e:
                        logger.warning(
                            f'Exception while trying to add KPI {kpi.name} to FeatureGroup {group_name}: {e}'
                        )
                    finally:
                        if self.include_related_kpis_in_tooltip:
                            if _tmp is not None:
                                generator.feature_resolver.property_mappers['tooltip'] = _tmp
                            else:
                                generator.feature_resolver.property_mappers.pop('tooltip')
                pbar.update(1)

            feature_groups.append(fg)

    return feature_groups

KPIGroupingManager

Manages sophisticated KPI grouping and organization for map visualization.

Handles the complex logic of grouping KPIs by their attributes, creating meaningful feature group names, and finding related KPIs for enhanced tooltips. Supports custom sorting orders and category hierarchies.

The grouping system is designed to create logical visual organization of energy system KPIs, where related metrics (same flag, different datasets) are grouped together and presented with consistent naming.

Parameters:

Name Type Description Default
kpi_attribute_category_orders dict[str, list[str]]

Custom ordering for specific attribute values

None
kpi_attribute_keys_to_exclude_from_grouping list[str]

Attributes to ignore during grouping

None
kpi_attribute_sort_order list[str]

Order of attributes for group sorting

None

Examples:

Custom grouping configuration:

>>> manager = KPIGroupingManager(
...     kpi_attribute_category_orders={
...         'dataset': ['reference', 'scenario_1', 'scenario_2'],
...         'aggregation': ['Sum', 'Mean', 'Max']
...     },
...     kpi_attribute_keys_to_exclude_from_grouping=['object_name']
... )
Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
class KPIGroupingManager:
    """
    Manages sophisticated KPI grouping and organization for map visualization.

    Handles the complex logic of grouping KPIs by their attributes, creating
    meaningful feature group names, and finding related KPIs for enhanced
    tooltips. Supports custom sorting orders and category hierarchies.

    The grouping system is designed to create logical visual organization
    of energy system KPIs, where related metrics (same flag, different datasets)
    are grouped together and presented with consistent naming.

    Args:
        kpi_attribute_category_orders: Custom ordering for specific attribute values
        kpi_attribute_keys_to_exclude_from_grouping: Attributes to ignore during grouping
        kpi_attribute_sort_order: Order of attributes for group sorting

    Examples:
        Custom grouping configuration:
        >>> manager = KPIGroupingManager(
        ...     kpi_attribute_category_orders={
        ...         'dataset': ['reference', 'scenario_1', 'scenario_2'],
        ...         'aggregation': ['Sum', 'Mean', 'Max']
        ...     },
        ...     kpi_attribute_keys_to_exclude_from_grouping=['object_name']
        ... )
    """

    DEFAULT_EXCLUDE_FROM_GROUPING = ['name', 'object_name', 'column_subset']
    DEFAULT_SORT_ORDER = [
        'name_prefix', 'model_flag', 'flag', 'model_query', 'aggregation',
        'reference_dataset', 'variation_dataset', 'dataset',
        'value_comparison', 'value_operation', 'name_suffix'
    ]
    DEFAULT_INCLUDE_ATTRIBUTES = ['value_operation', 'aggregation', 'flag', 'dataset', 'unit']
    DEFAULT_EXCLUDE_ATTRIBUTES = ['variation_dataset', 'reference_dataset', 'model_flag', 'base_unit', 'dataset_type']

    def __init__(
            self,
            kpi_attribute_category_orders: dict[str, list[str]] = None,
            kpi_attribute_keys_to_exclude_from_grouping: list[str] = None,
            kpi_attribute_sort_order: list[str] = None
    ):
        self.kpi_attribute_category_orders = kpi_attribute_category_orders or {}
        self.kpi_attribute_keys_to_exclude_from_grouping = (
                kpi_attribute_keys_to_exclude_from_grouping or self.DEFAULT_EXCLUDE_FROM_GROUPING.copy()
        )
        self.kpi_attribute_sort_order = (
                kpi_attribute_sort_order or self.DEFAULT_SORT_ORDER.copy()
        )

    def get_kpi_groups(self, kpi_collection: KPICollection) -> list[KPICollection]:
        """
        Group KPIs by attributes with sophisticated sorting.

        Creates logical groups of KPIs based on their attributes, excluding
        specified attributes from grouping and applying custom sort orders.

        Args:
            kpi_collection: Collection of KPIs to group

        Returns:
            List of KPICollection objects, each representing a logical group
        """
        from mesqual.utils.dict_combinations import dict_combination_iterator

        attribute_sets = kpi_collection.get_all_kpi_attributes_and_value_sets(primitive_values=True)
        relevant_attribute_sets = {
            k: v for k, v in attribute_sets.items()
            if k not in self.kpi_attribute_keys_to_exclude_from_grouping
        }

        ordered_keys = [k for k in self.kpi_attribute_sort_order if k in relevant_attribute_sets]

        # Build attribute value rankings
        attribute_value_rank: dict[str, dict[str, int]] = {}
        for attr in ordered_keys:
            existing_values = set(relevant_attribute_sets.get(attr, []))
            manual_order = [v for v in self.kpi_attribute_category_orders.get(attr, []) if v in existing_values]
            remaining = list(existing_values - set(manual_order))
            try:
                remaining = list(sorted(remaining))
            except TypeError:
                pass
            full_order = manual_order + remaining
            attribute_value_rank[attr] = {val: idx for idx, val in enumerate(full_order)}

        def sorting_index(group_kwargs: dict[str, str]) -> tuple:
            return tuple(
                attribute_value_rank[attr].get(group_kwargs.get(attr), float("inf"))
                for attr in ordered_keys
            )

        # Create and sort groups
        group_kwargs_list = list(dict_combination_iterator(relevant_attribute_sets))
        group_kwargs_list.sort(key=sorting_index)

        groups: list[KPICollection] = []
        for group_kwargs in group_kwargs_list:
            g = kpi_collection.get_filtered_kpi_collection_by_attributes(**group_kwargs)
            if not g.empty:
                groups.append(g)

        return groups

    def get_feature_group_name(self, kpi_group: KPICollection) -> str:
        """
        Generate meaningful feature group name from KPI group.

        Creates human-readable names for map feature groups based on
        common KPI attributes, prioritizing important attributes.

        Args:
            kpi_group: KPI group to generate name for

        Returns:
            Human-readable feature group name
        """
        attributes = kpi_group.get_in_common_kpi_attributes(primitive_values=True)

        for k in self.DEFAULT_EXCLUDE_ATTRIBUTES:
            attributes.pop(k, None)

        components = []
        include_attrs = self.DEFAULT_INCLUDE_ATTRIBUTES + [
            k for k in attributes.keys() if k not in self.DEFAULT_INCLUDE_ATTRIBUTES
        ]
        for k in include_attrs:
            value = attributes.pop(k, None)
            if value is not None:
                components.append(str(value))

        return ' '.join(components)

    def get_related_kpi_groups(self, kpi: KPI, study_manager) -> dict[str, KPICollection]:
        """
        Get related KPIs grouped by relationship type.

        Finds KPIs related to the given KPI across different dimensions
        (comparisons, aggregations, datasets) for enhanced tooltip display.

        Args:
            kpi: Source KPI to find relatives for
            study_manager: StudyManager for accessing related KPIs

        Returns:
            Dict mapping relationship type to KPICollection of related KPIs
        """
        from mesqual.kpis import ValueComparisonKPI, ArithmeticValueOperationKPI

        groups = {
            'Different Comparisons / ValueOperations': KPICollection(),
            'Different Aggregations': KPICollection(),
            'Different Datasets': KPICollection(),
        }

        if not study_manager:
            return groups

        kpi_atts = kpi.attributes.as_dict(primitive_values=True)

        _must_contain = ['flag', 'aggregation']
        if any(kpi_atts.get(k, None) is None for k in _must_contain):
            return groups

        try:
            pre_filtered = study_manager.scen_comp.get_merged_kpi_collection()
            pre_filtered = pre_filtered.get_filtered_kpi_collection_by_attributes(
                object_name=kpi.get_attributed_object_name(),
                flag=kpi_atts['flag'],
                model_flag=kpi.get_attributed_model_flag(),
            )
        except:
            return groups

        _main_kpi_is_value_op = isinstance(kpi, (ValueComparisonKPI, ArithmeticValueOperationKPI))

        for potential_relative in pre_filtered:
            pratts = potential_relative.attributes.as_dict(primitive_values=True)
            if pratts.get('dataset') == kpi_atts.get('dataset'):  # same ds
                if pratts.get('aggregation', None) == kpi_atts.get('aggregation'):  # same ds, agg
                    if pratts.get('value_operation', None) != kpi_atts.get('value_operation', None):
                        groups['Different Comparisons / ValueOperations'].add_kpi(potential_relative)
                        continue
                else:  # same ds, diff agg
                    if pratts.get('value_operation', None) is None:
                        groups['Different Aggregations'].add_kpi(potential_relative)
                        continue
                    elif pratts.get('value_operation') == kpi_atts.get('value_operation', None):
                        groups['Different Aggregations'].add_kpi(potential_relative)
                        continue
            elif pratts.get('aggregation', None) == kpi_atts.get('aggregation'):  # same agg, diff ds
                if pratts.get('value_operation', None) == kpi_atts.get('value_operation', None):
                    groups['Different Datasets'].add_kpi(potential_relative)
                    continue
                if not _main_kpi_is_value_op:
                    groups['Different Comparisons / ValueOperations'].add_kpi(potential_relative)
                    continue

        return groups

get_kpi_groups

get_kpi_groups(kpi_collection: KPICollection) -> list[KPICollection]

Group KPIs by attributes with sophisticated sorting.

Creates logical groups of KPIs based on their attributes, excluding specified attributes from grouping and applying custom sort orders.

Parameters:

Name Type Description Default
kpi_collection KPICollection

Collection of KPIs to group

required

Returns:

Type Description
list[KPICollection]

List of KPICollection objects, each representing a logical group

Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def get_kpi_groups(self, kpi_collection: KPICollection) -> list[KPICollection]:
    """
    Group KPIs by attributes with sophisticated sorting.

    Creates logical groups of KPIs based on their attributes, excluding
    specified attributes from grouping and applying custom sort orders.

    Args:
        kpi_collection: Collection of KPIs to group

    Returns:
        List of KPICollection objects, each representing a logical group
    """
    from mesqual.utils.dict_combinations import dict_combination_iterator

    attribute_sets = kpi_collection.get_all_kpi_attributes_and_value_sets(primitive_values=True)
    relevant_attribute_sets = {
        k: v for k, v in attribute_sets.items()
        if k not in self.kpi_attribute_keys_to_exclude_from_grouping
    }

    ordered_keys = [k for k in self.kpi_attribute_sort_order if k in relevant_attribute_sets]

    # Build attribute value rankings
    attribute_value_rank: dict[str, dict[str, int]] = {}
    for attr in ordered_keys:
        existing_values = set(relevant_attribute_sets.get(attr, []))
        manual_order = [v for v in self.kpi_attribute_category_orders.get(attr, []) if v in existing_values]
        remaining = list(existing_values - set(manual_order))
        try:
            remaining = list(sorted(remaining))
        except TypeError:
            pass
        full_order = manual_order + remaining
        attribute_value_rank[attr] = {val: idx for idx, val in enumerate(full_order)}

    def sorting_index(group_kwargs: dict[str, str]) -> tuple:
        return tuple(
            attribute_value_rank[attr].get(group_kwargs.get(attr), float("inf"))
            for attr in ordered_keys
        )

    # Create and sort groups
    group_kwargs_list = list(dict_combination_iterator(relevant_attribute_sets))
    group_kwargs_list.sort(key=sorting_index)

    groups: list[KPICollection] = []
    for group_kwargs in group_kwargs_list:
        g = kpi_collection.get_filtered_kpi_collection_by_attributes(**group_kwargs)
        if not g.empty:
            groups.append(g)

    return groups

get_feature_group_name

get_feature_group_name(kpi_group: KPICollection) -> str

Generate meaningful feature group name from KPI group.

Creates human-readable names for map feature groups based on common KPI attributes, prioritizing important attributes.

Parameters:

Name Type Description Default
kpi_group KPICollection

KPI group to generate name for

required

Returns:

Type Description
str

Human-readable feature group name

Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def get_feature_group_name(self, kpi_group: KPICollection) -> str:
    """
    Generate meaningful feature group name from KPI group.

    Creates human-readable names for map feature groups based on
    common KPI attributes, prioritizing important attributes.

    Args:
        kpi_group: KPI group to generate name for

    Returns:
        Human-readable feature group name
    """
    attributes = kpi_group.get_in_common_kpi_attributes(primitive_values=True)

    for k in self.DEFAULT_EXCLUDE_ATTRIBUTES:
        attributes.pop(k, None)

    components = []
    include_attrs = self.DEFAULT_INCLUDE_ATTRIBUTES + [
        k for k in attributes.keys() if k not in self.DEFAULT_INCLUDE_ATTRIBUTES
    ]
    for k in include_attrs:
        value = attributes.pop(k, None)
        if value is not None:
            components.append(str(value))

    return ' '.join(components)
get_related_kpi_groups(kpi: KPI, study_manager) -> dict[str, KPICollection]

Get related KPIs grouped by relationship type.

Finds KPIs related to the given KPI across different dimensions (comparisons, aggregations, datasets) for enhanced tooltip display.

Parameters:

Name Type Description Default
kpi KPI

Source KPI to find relatives for

required
study_manager

StudyManager for accessing related KPIs

required

Returns:

Type Description
dict[str, KPICollection]

Dict mapping relationship type to KPICollection of related KPIs

Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def get_related_kpi_groups(self, kpi: KPI, study_manager) -> dict[str, KPICollection]:
    """
    Get related KPIs grouped by relationship type.

    Finds KPIs related to the given KPI across different dimensions
    (comparisons, aggregations, datasets) for enhanced tooltip display.

    Args:
        kpi: Source KPI to find relatives for
        study_manager: StudyManager for accessing related KPIs

    Returns:
        Dict mapping relationship type to KPICollection of related KPIs
    """
    from mesqual.kpis import ValueComparisonKPI, ArithmeticValueOperationKPI

    groups = {
        'Different Comparisons / ValueOperations': KPICollection(),
        'Different Aggregations': KPICollection(),
        'Different Datasets': KPICollection(),
    }

    if not study_manager:
        return groups

    kpi_atts = kpi.attributes.as_dict(primitive_values=True)

    _must_contain = ['flag', 'aggregation']
    if any(kpi_atts.get(k, None) is None for k in _must_contain):
        return groups

    try:
        pre_filtered = study_manager.scen_comp.get_merged_kpi_collection()
        pre_filtered = pre_filtered.get_filtered_kpi_collection_by_attributes(
            object_name=kpi.get_attributed_object_name(),
            flag=kpi_atts['flag'],
            model_flag=kpi.get_attributed_model_flag(),
        )
    except:
        return groups

    _main_kpi_is_value_op = isinstance(kpi, (ValueComparisonKPI, ArithmeticValueOperationKPI))

    for potential_relative in pre_filtered:
        pratts = potential_relative.attributes.as_dict(primitive_values=True)
        if pratts.get('dataset') == kpi_atts.get('dataset'):  # same ds
            if pratts.get('aggregation', None) == kpi_atts.get('aggregation'):  # same ds, agg
                if pratts.get('value_operation', None) != kpi_atts.get('value_operation', None):
                    groups['Different Comparisons / ValueOperations'].add_kpi(potential_relative)
                    continue
            else:  # same ds, diff agg
                if pratts.get('value_operation', None) is None:
                    groups['Different Aggregations'].add_kpi(potential_relative)
                    continue
                elif pratts.get('value_operation') == kpi_atts.get('value_operation', None):
                    groups['Different Aggregations'].add_kpi(potential_relative)
                    continue
        elif pratts.get('aggregation', None) == kpi_atts.get('aggregation'):  # same agg, diff ds
            if pratts.get('value_operation', None) == kpi_atts.get('value_operation', None):
                groups['Different Datasets'].add_kpi(potential_relative)
                continue
            if not _main_kpi_is_value_op:
                groups['Different Comparisons / ValueOperations'].add_kpi(potential_relative)
                continue

    return groups