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.

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(folium_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'
... )
Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
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
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
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.

    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(folium_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'
        ... )
    """

    def __init__(
            self,
            generators: FoliumObjectGenerator | List[FoliumObjectGenerator],
            study_manager: 'StudyManager' = None,
            include_related_kpis_in_tooltip: bool = False,
            kpi_grouping_manager: KPIGroupingManager = None,
            value_formatting: VALUE_FORMATTING_OPTIONS = 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.value_formatting = value_formatting

        self.grouping_manager = kpi_grouping_manager or KPIGroupingManager()
        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]:
        """Generate feature groups and add them to map."""
        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
        """
        from tqdm import tqdm
        from mesqual.utils.logging import get_logger

        logger = get_logger(__name__)
        feature_groups = []
        failed = []

        kpi_to_text_converter_resolver = KPIQuantityToTextConverterResolver(self.value_formatting, kpi_collection)

        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)
                converter = kpi_to_text_converter_resolver.get_converter_for_kpi(group_name, kpi_group)

                _generators = [
                    self._create_generator_with_updated_converters_if_needed(converter, generator)
                    for generator in self.generators
                ]

                for kpi in kpi_group:
                    data_item = KPIDataItem(kpi, kpi_collection, study_manager=self.study_manager, **self.kwargs)
                    for gen in _generators:
                        try:
                            gen.generate(data_item, fg)
                        except Exception as e:
                            failed.append((kpi.name, group_name, e))

                    pbar.update(1)

                feature_groups.append(fg)

        if failed:
            logger.warning(
                f'Exception while trying to add {len(failed)} KPIs: {failed[:3]}'
            )
        return feature_groups

    def _create_generator_with_updated_converters_if_needed(self, converter, generator):
        """Create generator copy with converter-specific mappers if needed."""
        import copy

        # Check if we need to modify any mappers
        needs_modification = (
            self.include_related_kpis_in_tooltip or
            'text_print_content' in generator.feature_resolver.property_mappers
        )

        if not needs_modification:
            return generator

        # Create shallow copies
        gen = copy.copy(generator)
        gen.feature_resolver = copy.copy(generator.feature_resolver)
        gen.feature_resolver.property_mappers = generator.feature_resolver.property_mappers.copy()

        if self.include_related_kpis_in_tooltip:
            gen.feature_resolver.property_mappers['tooltip'] = self._create_enhanced_tooltip_generator(converter)

        if (self.value_formatting is not None) and ('text_print_content' in gen.feature_resolver.property_mappers):
            gen.feature_resolver.property_mappers['text_print_content'] = self._create_text_print_generator(converter)

        return gen

    def _create_text_print_generator(self, converter: QuantityToTextConverter) -> PropertyMapper:
        """
        Create text print generator with fixed converter.

        Args:
            converter: QuantityToTextConverter to use for formatting

        Returns:
            PropertyMapper with converter baked into the closure
        """
        def get_text(data_item: VisualizableDataItem) -> str:
            return data_item.get_text_representation(converter)

        return PropertyMapper(get_text)

    @staticmethod
    def _create_default_tooltip_generator(converter: QuantityToTextConverter) -> PropertyMapper:
        """
        Create default tooltip generator showing data item information.

        Args:
            converter: QuantityToTextConverter to use for formatting all values

        Returns:
            PropertyMapper that generates HTML table tooltips with data item attributes
        """

        def get_tooltip(data_item: VisualizableDataItem) -> str:
            tooltip_data = data_item.get_tooltip_data(converter=converter)

            html = '<table style="border-collapse: collapse;">\n'
            for key, value in tooltip_data.items():
                html += f'  <tr><td style="padding: 4px 8px;"><strong>{key}</strong></td>' \
                        f'<td style="text-align: right; padding: 4px 8px;">{value}</td></tr>\n'
            html += '</table>'

            return html

        return PropertyMapper(get_tooltip)

    def _create_enhanced_tooltip_generator(self, converter: QuantityToTextConverter) -> PropertyMapper:
        """
        Create tooltip generator that includes related KPIs with fixed converter.

        Args:
            converter: QuantityToTextConverter to use for formatting all values

        Returns:
            PropertyMapper with converter baked into the closure
        """

        def generate_tooltip(data_item: KPIDataItem) -> str:
            kpi = data_item.kpi
            kpi_name = kpi.get_kpi_name_with_dataset_name()

            kpi_text = converter.convert(kpi.quantity)

            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_value_text = converter.convert(related_kpi.quantity)
                            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)

generate_and_add_feature_groups_to_map

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

Generate feature groups and add them to map.

Source code in submodules/mesqual/mesqual/visualizations/folium_viz_system/kpi_collection_map_visualizer.py
400
401
402
403
404
405
406
407
408
409
410
411
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]:
    """Generate feature groups and add them to map."""
    fgs = self.get_feature_groups(kpi_collection, show=show, overlay=overlay)
    for fg in fgs:
        folium_map.add_child(fg)
    return fgs

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
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
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
    """
    from tqdm import tqdm
    from mesqual.utils.logging import get_logger

    logger = get_logger(__name__)
    feature_groups = []
    failed = []

    kpi_to_text_converter_resolver = KPIQuantityToTextConverterResolver(self.value_formatting, kpi_collection)

    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)
            converter = kpi_to_text_converter_resolver.get_converter_for_kpi(group_name, kpi_group)

            _generators = [
                self._create_generator_with_updated_converters_if_needed(converter, generator)
                for generator in self.generators
            ]

            for kpi in kpi_group:
                data_item = KPIDataItem(kpi, kpi_collection, study_manager=self.study_manager, **self.kwargs)
                for gen in _generators:
                    try:
                        gen.generate(data_item, fg)
                    except Exception as e:
                        failed.append((kpi.name, group_name, e))

                pbar.update(1)

            feature_groups.append(fg)

    if failed:
        logger.warning(
            f'Exception while trying to add {len(failed)} KPIs: {failed[:3]}'
        )
    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_name': ['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
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
216
217
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
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_name': ['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', 'custom_name',
                                      'name_prefix', 'name_suffix', 'unit', 'target_unit']
    DEFAULT_SORT_ORDER = [
        'name_prefix', 'model_flag', 'flag', 'aggregation',
        'reference_dataset_name', 'variation_dataset_name', 'dataset_name',
        'value_comparison', 'arithmetic_operation', 'name_suffix'
    ]
    DEFAULT_INCLUDE_ATTRIBUTES = ['arithmetic_operation', 'aggregation', 'flag', 'dataset_name', 'unit']
    DEFAULT_EXCLUDE_ATTRIBUTES = ['variation_dataset_name', 'reference_dataset_name', 'model_flag',
                                   'dataset_type', 'target_unit', 'dataset_attributes']

    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.filter(**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 (FIXED VERSION).

        Finds KPIs related to the given KPI across different dimensions:
        - Same object/flag, different aggregations
        - Same object/flag/aggregation, different datasets
        - Same object/flag/aggregation, different comparisons/operations

        This is a corrected version that properly separates different relationship types.

        Args:
            kpi: Source KPI to find relatives for
            study_manager: StudyManager for accessing merged KPI collection

        Returns:
            Dict mapping relationship type to KPICollection of related KPIs
        """
        groups = {
            'Different Comparisons / Operations': KPICollection(),
            'Different Aggregations': KPICollection(),
            'Different Datasets': KPICollection(),
        }

        if not study_manager:
            return groups

        # Get reference KPI attributes
        object_name = kpi.attributes.object_name
        flag = kpi.attributes.flag
        model_flag = kpi.attributes.model_flag
        aggregation = kpi.attributes.aggregation
        dataset_name = kpi.attributes.dataset_name
        value_comparison = kpi.attributes.value_comparison
        arithmetic_operation = kpi.attributes.arithmetic_operation

        if not flag or aggregation is None:
            return groups

        try:
            # Get all KPIs for same object/flag/model_flag
            all_related = study_manager.scen_comp.get_merged_kpi_collection()
            pre_filtered = all_related.filter(
                object_name=object_name,
                flag=flag,
                model_flag=model_flag
            )
        except:
            return groups

        # Determine if main KPI has comparison/operation
        main_has_comparison = value_comparison is not None or arithmetic_operation is not None

        for potential in pre_filtered:
            # Skip self
            if potential is kpi:
                continue

            pot_agg = potential.attributes.aggregation
            pot_dataset = potential.attributes.dataset_name
            pot_comparison = potential.attributes.value_comparison
            pot_operation = potential.attributes.arithmetic_operation
            pot_has_comparison = pot_comparison is not None or pot_operation is not None

            # Category 1: Different Aggregations
            # Same dataset, same comparison status, different aggregation
            if (pot_dataset == dataset_name and
                pot_agg != aggregation and
                pot_comparison == value_comparison and
                pot_operation == arithmetic_operation):
                groups['Different Aggregations'].add(potential)
                continue

            # Category 2: Different Datasets
            # Same aggregation, same comparison status, different dataset
            if (pot_agg == aggregation and
                pot_dataset != dataset_name and
                pot_comparison == value_comparison and
                pot_operation == arithmetic_operation):
                groups['Different Datasets'].add(potential)
                continue

            # Category 3: Different Comparisons/Operations
            # Same dataset, same aggregation, different comparison/operation
            if (pot_dataset == dataset_name and
                pot_agg == aggregation and
                (pot_comparison != value_comparison or pot_operation != arithmetic_operation)):
                groups['Different Comparisons / Operations'].add(potential)
                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
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
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.filter(**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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
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 (FIXED VERSION).

Finds KPIs related to the given KPI across different dimensions: - Same object/flag, different aggregations - Same object/flag/aggregation, different datasets - Same object/flag/aggregation, different comparisons/operations

This is a corrected version that properly separates different relationship types.

Parameters:

Name Type Description Default
kpi KPI

Source KPI to find relatives for

required
study_manager

StudyManager for accessing merged KPI collection

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
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
def get_related_kpi_groups(self, kpi: KPI, study_manager) -> dict[str, KPICollection]:
    """
    Get related KPIs grouped by relationship type (FIXED VERSION).

    Finds KPIs related to the given KPI across different dimensions:
    - Same object/flag, different aggregations
    - Same object/flag/aggregation, different datasets
    - Same object/flag/aggregation, different comparisons/operations

    This is a corrected version that properly separates different relationship types.

    Args:
        kpi: Source KPI to find relatives for
        study_manager: StudyManager for accessing merged KPI collection

    Returns:
        Dict mapping relationship type to KPICollection of related KPIs
    """
    groups = {
        'Different Comparisons / Operations': KPICollection(),
        'Different Aggregations': KPICollection(),
        'Different Datasets': KPICollection(),
    }

    if not study_manager:
        return groups

    # Get reference KPI attributes
    object_name = kpi.attributes.object_name
    flag = kpi.attributes.flag
    model_flag = kpi.attributes.model_flag
    aggregation = kpi.attributes.aggregation
    dataset_name = kpi.attributes.dataset_name
    value_comparison = kpi.attributes.value_comparison
    arithmetic_operation = kpi.attributes.arithmetic_operation

    if not flag or aggregation is None:
        return groups

    try:
        # Get all KPIs for same object/flag/model_flag
        all_related = study_manager.scen_comp.get_merged_kpi_collection()
        pre_filtered = all_related.filter(
            object_name=object_name,
            flag=flag,
            model_flag=model_flag
        )
    except:
        return groups

    # Determine if main KPI has comparison/operation
    main_has_comparison = value_comparison is not None or arithmetic_operation is not None

    for potential in pre_filtered:
        # Skip self
        if potential is kpi:
            continue

        pot_agg = potential.attributes.aggregation
        pot_dataset = potential.attributes.dataset_name
        pot_comparison = potential.attributes.value_comparison
        pot_operation = potential.attributes.arithmetic_operation
        pot_has_comparison = pot_comparison is not None or pot_operation is not None

        # Category 1: Different Aggregations
        # Same dataset, same comparison status, different aggregation
        if (pot_dataset == dataset_name and
            pot_agg != aggregation and
            pot_comparison == value_comparison and
            pot_operation == arithmetic_operation):
            groups['Different Aggregations'].add(potential)
            continue

        # Category 2: Different Datasets
        # Same aggregation, same comparison status, different dataset
        if (pot_agg == aggregation and
            pot_dataset != dataset_name and
            pot_comparison == value_comparison and
            pot_operation == arithmetic_operation):
            groups['Different Datasets'].add(potential)
            continue

        # Category 3: Different Comparisons/Operations
        # Same dataset, same aggregation, different comparison/operation
        if (pot_dataset == dataset_name and
            pot_agg == aggregation and
            (pot_comparison != value_comparison or pot_operation != arithmetic_operation)):
            groups['Different Comparisons / Operations'].add(potential)
            continue

    return groups