MESQUAL 201: KPI Framework and Unit Handling¶

This notebook demonstrates how to build a structured KPI (Key Performance Indicator) system with proper unit handling using MESQUAL's new KPI framework.

Introduction¶

MESQUAL's KPI framework provides a sophisticated system for creating, managing, and analyzing performance metrics across energy scenarios. This notebook covers:

1. KPI Builder Pattern: Declarative KPI creation with fluent API
2. Batch Generation: Efficient computation across multiple objects
3. Unit Handling: Automatic unit conversion and pretty formatting
4. Aggregations: Various statistical aggregations (Sum, Mean, Max, etc.)
5. KPI Attributes: Rich metadata for filtering and organization
6. Comparison KPIs: Scenario comparison workflows

The new KPI system is designed for 10x performance improvement through batch-first architecture while providing a cleaner, more intuitive API.

Setup¶

First, we need to set up the environment. If you are on Colab, the first cell will clone and install all dependencies. You will have to restart the session afterwards and continue with cell 2. If you are in a local environment, make sure that you have followed the Getting started steps in the README, so that mesqual and all requirements are installed.

In [1]:

import os

if "COLAB_RELEASE_TAG" in os.environ:
    import importlib.util

    def is_module_available(module_name):
        return importlib.util.find_spec(module_name) is not None

    if os.path.exists("mesqual-vanilla-studies") and is_module_available("mesqual"):
        print("✅ Environment already set up. Skipping installation.")
    else:
        print("🔧 Setting up Colab environment...")
        !git clone --recursive https://github.com/helgeesch/mesqual-vanilla-studies.git
        %cd mesqual-vanilla-studies/

        !pip install git+https://github.com/helgeesch/mesqual -U
        !pip install git+https://github.com/helgeesch/mesqual-pypsa -U
        !pip install -r requirements.txt

        print('✅ Setup complete. 🔁 Restart the session, then skip this cell and continue with the next one.')
else:
    print("🖥️ Running locally. No setup needed.")

import os if "COLAB_RELEASE_TAG" in os.environ: import importlib.util def is_module_available(module_name): return importlib.util.find_spec(module_name) is not None if os.path.exists("mesqual-vanilla-studies") and is_module_available("mesqual"): print("✅ Environment already set up. Skipping installation.") else: print("🔧 Setting up Colab environment...") !git clone --recursive https://github.com/helgeesch/mesqual-vanilla-studies.git %cd mesqual-vanilla-studies/ !pip install git+https://github.com/helgeesch/mesqual -U !pip install git+https://github.com/helgeesch/mesqual-pypsa -U !pip install -r requirements.txt print('✅ Setup complete. 🔁 Restart the session, then skip this cell and continue with the next one.') else: print("🖥️ Running locally. No setup needed.")

🖥️ Running locally. No setup needed.

In [2]:

import os

if "COLAB_RELEASE_TAG" in os.environ:
    import sys
    sys.path.append('/content/mesqual-vanilla-studies')
    os.chdir('/content/mesqual-vanilla-studies')
else:
    def setup_notebook_env():
        """Set working directory to repo root and ensure it's in sys.path."""
        import os
        import sys
        from pathlib import Path

        def find_repo_root(start_path: Path) -> Path:
            current = start_path.resolve()
            while current != current.parent:
                if (current / 'vanilla').exists():
                    return current
                current = current.parent
            raise FileNotFoundError(f"Repository root not found from: {start_path}")

        repo_root = find_repo_root(Path.cwd())
        os.chdir(repo_root)
        if str(repo_root) not in sys.path:
            sys.path.insert(0, str(repo_root))

    setup_notebook_env()

try:
    from mesqual import StudyManager
except ImportError:
    raise ImportError("❌ 'mesqual' not found. If you're running locally, make sure you've installed all dependencies as described in the README.")

if not os.path.isdir("studies"):
    raise RuntimeError(f"❌ 'studies' folder not found. Make sure your working directory is set to the mesqual-vanilla-studies root. Current working directory: {os.getcwd()}")

print("✅ Environment ready. Let's go!")

import os if "COLAB_RELEASE_TAG" in os.environ: import sys sys.path.append('/content/mesqual-vanilla-studies') os.chdir('/content/mesqual-vanilla-studies') else: def setup_notebook_env(): """Set working directory to repo root and ensure it's in sys.path.""" import os import sys from pathlib import Path def find_repo_root(start_path: Path) -> Path: current = start_path.resolve() while current != current.parent: if (current / 'vanilla').exists(): return current current = current.parent raise FileNotFoundError(f"Repository root not found from: {start_path}") repo_root = find_repo_root(Path.cwd()) os.chdir(repo_root) if str(repo_root) not in sys.path: sys.path.insert(0, str(repo_root)) setup_notebook_env() try: from mesqual import StudyManager except ImportError: raise ImportError("❌ 'mesqual' not found. If you're running locally, make sure you've installed all dependencies as described in the README.") if not os.path.isdir("studies"): raise RuntimeError(f"❌ 'studies' folder not found. Make sure your working directory is set to the mesqual-vanilla-studies root. Current working directory: {os.getcwd()}") print("✅ Environment ready. Let's go!")

✅ Environment ready. Let's go!

In [3]:

import pandas as pd
from mesqual import kpis
from mesqual.units import Units
from vanilla.notebook_config import configure_clean_output_for_jupyter_notebook

configure_clean_output_for_jupyter_notebook()

import pandas as pd from mesqual import kpis from mesqual.units import Units from vanilla.notebook_config import configure_clean_output_for_jupyter_notebook configure_clean_output_for_jupyter_notebook()

Load Study Data¶

Let's load our study data using the same Scigrid-DE setup:

In [4]:

from studies.study_01_intro_to_mesqual.scripts.setup_study_manager import get_scigrid_de_study_manager

study = get_scigrid_de_study_manager()

print("Study scenarios:")
for dataset in study.scen.datasets:
    print(f"  📊 {dataset.name}")

# Get base dataset for exploration
ds_base = study.scen.get_dataset('base')
print(f"\n✅ Loaded dataset: {ds_base.name}")

from studies.study_01_intro_to_mesqual.scripts.setup_study_manager import get_scigrid_de_study_manager study = get_scigrid_de_study_manager() print("Study scenarios:") for dataset in study.scen.datasets: print(f" 📊 {dataset.name}") # Get base dataset for exploration ds_base = study.scen.get_dataset('base') print(f"\n✅ Loaded dataset: {ds_base.name}")

Study scenarios:
  📊 base
  📊 solar_150
  📊 solar_200
  📊 wind_150
  📊 wind_200

✅ Loaded dataset: base

Part 1: Building KPIs with the Builder Pattern¶

The KPI framework uses a fluent builder pattern for declarative KPI creation:

Creating a Simple KPI¶

Let's create a KPI for average market price in a control area:

In [5]:

# Define KPI using builder pattern
price_kpi_definition = (
    kpis.FlagAggKPIBuilder()
    .for_flag('control_areas_t.vol_weighted_marginal_price')
    .with_aggregation(kpis.Aggregations.Mean)
    .for_objects(['TenneTDE', '50Hertz'])
    .build()
)

# Generate KPI
ds_base.add_kpis_from_definitions(price_kpi_definition)

# Examine the KPI
kpi = ds_base.kpi_collection._kpis[0]
print(f"KPI Name: {kpi.name}")
print(f"Value: {kpi.value:.2f}")
print(f"Quantity: {kpi.quantity}")
print(f"\nKPI Attributes:")
for attr, value in kpi.attributes.as_dict(primitive_values=True).items():
    print(f"  {attr}: {value}")

# Define KPI using builder pattern price_kpi_definition = ( kpis.FlagAggKPIBuilder() .for_flag('control_areas_t.vol_weighted_marginal_price') .with_aggregation(kpis.Aggregations.Mean) .for_objects(['TenneTDE', '50Hertz']) .build() ) # Generate KPI ds_base.add_kpis_from_definitions(price_kpi_definition) # Examine the KPI kpi = ds_base.kpi_collection._kpis[0] print(f"KPI Name: {kpi.name}") print(f"Value: {kpi.value:.2f}") print(f"Quantity: {kpi.quantity}") print(f"\nKPI Attributes:") for attr, value in kpi.attributes.as_dict(primitive_values=True).items(): print(f" {attr}: {value}")

KPI Name: control_areas_t.vol_weighted_marginal_price Mean TenneTDE
Value: 16.91
Quantity: 16.906856836420513 EUR_per_MWh

KPI Attributes:
  flag: control_areas_t.vol_weighted_marginal_price
  model_flag: control_areas
  object_name: TenneTDE
  aggregation: Mean
  dataset_name: base
  dataset_type: <class 'studies.study_01_intro_to_mesqual.scripts.setup_study_manager.ScigridDEDataset'>
  value_comparison: None
  arithmetic_operation: None
  reference_dataset_name: None
  variation_dataset_name: None
  name_prefix: 
  name_suffix: 
  custom_name: None
  unit: EUR_per_MWh

Batch KPI Generation¶

The real power comes from batch generation across multiple objects:

In [6]:

# Get all control areas
control_areas = ds_base.fetch('control_areas').index.to_list()
print(f"Control areas: {control_areas}")

# Create KPIs for all control areas at once
all_price_kpis_def = (
    kpis.FlagAggKPIBuilder()
    .for_flag('control_areas_t.vol_weighted_marginal_price')
    .with_aggregation(kpis.Aggregations.Mean)
    # .for_all_objects()  # is the default, can be skipped
    .build()
)

ds_base.add_kpis_from_definitions(all_price_kpis_def)

print(f"\n✅ Generated {len(all_price_kpis_def)} KPIs in one operation")
print("\nSample KPIs:")
for kpi in ds_base.kpi_collection[:20]:
    print(f"  {kpi.name}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

# Get all control areas control_areas = ds_base.fetch('control_areas').index.to_list() print(f"Control areas: {control_areas}") # Create KPIs for all control areas at once all_price_kpis_def = ( kpis.FlagAggKPIBuilder() .for_flag('control_areas_t.vol_weighted_marginal_price') .with_aggregation(kpis.Aggregations.Mean) # .for_all_objects() # is the default, can be skipped .build() ) ds_base.add_kpis_from_definitions(all_price_kpis_def) print(f"\n✅ Generated {len(all_price_kpis_def)} KPIs in one operation") print("\nSample KPIs:") for kpi in ds_base.kpi_collection[:20]: print(f" {kpi.name}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

Control areas: ['50Hertz', 'TenneTDE', 'TransnetBW', 'Amprion']

✅ Generated 1 KPIs in one operation

Sample KPIs:
  control_areas_t.vol_weighted_marginal_price Mean TenneTDE: 16.9 €/MWh
  control_areas_t.vol_weighted_marginal_price Mean 50Hertz: 11.9 €/MWh
  control_areas_t.vol_weighted_marginal_price Mean Amprion: 20.3 €/MWh
  control_areas_t.vol_weighted_marginal_price Mean TransnetBW: 23.6 €/MWh

Part 2: Aggregation Types¶

MESQUAL supports various aggregation types:

In [7]:

# Create KPIs with different kpis.Aggregations
kpi_defs = (
    kpis.FlagAggKPIBuilder()
    .for_flag('control_areas_t.vol_weighted_marginal_price')
    .with_aggregations([kpis.Aggregations.Sum, kpis.Aggregations.Max, kpis.Aggregations.Min])
    .for_objects(['TenneTDE', '50Hertz'])
    .build()
)
ds_base.add_kpis_from_definitions(kpi_defs)
kpi = ds_base.kpi_collection[-1]
print(f"{kpi.name:10s}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

# Create KPIs with different kpis.Aggregations kpi_defs = ( kpis.FlagAggKPIBuilder() .for_flag('control_areas_t.vol_weighted_marginal_price') .with_aggregations([kpis.Aggregations.Sum, kpis.Aggregations.Max, kpis.Aggregations.Min]) .for_objects(['TenneTDE', '50Hertz']) .build() ) ds_base.add_kpis_from_definitions(kpi_defs) kpi = ds_base.kpi_collection[-1] print(f"{kpi.name:10s}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

control_areas_t.vol_weighted_marginal_price Min 50Hertz: 6.76 €/MWh

Part 3: Unit Handling¶

MESQUAL's unit system provides automatic conversion and pretty formatting:

Automatic Pretty Units¶

In [8]:

# Create generation KPI
gen_kpi_def = (
    kpis.FlagAggKPIBuilder()
    .for_flag('generators_t.p')
    .with_aggregation(kpis.Aggregations.Mean)
    .build()
)

ds_base.add_kpis_from_definitions(gen_kpi_def)

# Show original and pretty units
sample_kpi = ds_base.kpi_collection[-1]
print(f"Original quantity: {sample_kpi.quantity}")
print(f"Pretty unit: {Units.get_quantity_in_pretty_unit(sample_kpi.quantity)}")
print(f"Pretty text: {Units.get_pretty_text_for_quantity(Units.get_quantity_in_pretty_unit(sample_kpi.quantity), thousands_separator=' ')}")

# Create generation KPI gen_kpi_def = ( kpis.FlagAggKPIBuilder() .for_flag('generators_t.p') .with_aggregation(kpis.Aggregations.Mean) .build() ) ds_base.add_kpis_from_definitions(gen_kpi_def) # Show original and pretty units sample_kpi = ds_base.kpi_collection[-1] print(f"Original quantity: {sample_kpi.quantity}") print(f"Pretty unit: {Units.get_quantity_in_pretty_unit(sample_kpi.quantity)}") print(f"Pretty text: {Units.get_pretty_text_for_quantity(Units.get_quantity_in_pretty_unit(sample_kpi.quantity), thousands_separator=' ')}")

Original quantity: 0.7883543532348286 MW
Pretty unit: 788.3543532348286 kW
Pretty text: 788 kW

KPI Unit Conversion Methods¶

In [9]:

# Get a KPI with energy units
print("Original KPI:")
print(f"  Value: {sample_kpi.value:.2f}")
print(f"  Unit: {sample_kpi.attributes.unit}")
print(f"  Quantity: {sample_kpi.quantity}")

# Convert to different unit
quantity_in_GW = Units.get_quantity_in_target_unit(sample_kpi.quantity, Units.GW)
GW_text = Units.get_pretty_text_for_quantity(quantity_in_GW)
print(f"\nConverted to GW: {GW_text}")

# Get in pretty unit
quantity_pretty = Units.get_quantity_in_pretty_unit(quantity_in_GW)
text_pretty = Units.get_pretty_text_for_quantity(quantity_pretty)
print(f"\nPretty unit quantity: {text_pretty}")

# Get a KPI with energy units print("Original KPI:") print(f" Value: {sample_kpi.value:.2f}") print(f" Unit: {sample_kpi.attributes.unit}") print(f" Quantity: {sample_kpi.quantity}") # Convert to different unit quantity_in_GW = Units.get_quantity_in_target_unit(sample_kpi.quantity, Units.GW) GW_text = Units.get_pretty_text_for_quantity(quantity_in_GW) print(f"\nConverted to GW: {GW_text}") # Get in pretty unit quantity_pretty = Units.get_quantity_in_pretty_unit(quantity_in_GW) text_pretty = Units.get_pretty_text_for_quantity(quantity_pretty) print(f"\nPretty unit quantity: {text_pretty}")

Original KPI:
  Value: 0.79
  Unit: MW
  Quantity: 0.7883543532348286 MW

Converted to GW: 0.00079 GW

Pretty unit quantity: 788 kW

Part 4: KPI Attributes and Metadata¶

Every KPI carries rich metadata for filtering and organization:

In [10]:

# Create KPI with custom attributes
kpi_def = (
    kpis.FlagAggKPIBuilder()
    .for_flag('control_areas_t.vol_weighted_marginal_price')
    .with_aggregation(kpis.Aggregations.Mean)
    .for_objects(['TenneTDE', '50Hertz'])
    .with_custom_name('Price')
    .with_extra_attributes(
        custom_category='custom value 123'
    )
    .build()
)
ds_base.add_kpis_from_definitions(kpi_def)

# Show original and pretty units
sample_kpi = ds_base.kpi_collection[0]

print("KPI Attributes:")
attrs = kpi.attributes.as_dict(primitive_values=True)
for key, value in sorted(attrs.items()):
    if value is not None:
        print(f"  {key:20s}: {value}")

# Create KPI with custom attributes kpi_def = ( kpis.FlagAggKPIBuilder() .for_flag('control_areas_t.vol_weighted_marginal_price') .with_aggregation(kpis.Aggregations.Mean) .for_objects(['TenneTDE', '50Hertz']) .with_custom_name('Price') .with_extra_attributes( custom_category='custom value 123' ) .build() ) ds_base.add_kpis_from_definitions(kpi_def) # Show original and pretty units sample_kpi = ds_base.kpi_collection[0] print("KPI Attributes:") attrs = kpi.attributes.as_dict(primitive_values=True) for key, value in sorted(attrs.items()): if value is not None: print(f" {key:20s}: {value}")

KPI Attributes:
  aggregation         : Min
  dataset_name        : base
  dataset_type        : <class 'studies.study_01_intro_to_mesqual.scripts.setup_study_manager.ScigridDEDataset'>
  flag                : control_areas_t.vol_weighted_marginal_price
  model_flag          : control_areas
  name_prefix         : 
  name_suffix         : 
  object_name         : 50Hertz
  unit                : EUR_per_MWh

Part 5: Multi-Scenario KPI Generation¶

Generate KPIs across multiple scenarios efficiently:

In [11]:

# Clear existing KPIs
study.scen.clear_kpi_collection_for_all_child_datasets()

# Generate price KPIs for all scenarios
price_def = (
    kpis.FlagAggKPIBuilder()
    .for_flag('control_areas_t.vol_weighted_marginal_price')
    .with_aggregation(kpis.Aggregations.Mean)
    .for_all_objects()
    .build()
)

for dataset in study.scen.datasets:
    dataset.add_kpis_from_definitions(price_def)
    print(f"✅ Generated {len(price_def)} KPIs for {dataset.name}")

# Get merged collection
all_kpis = study.scen.get_merged_kpi_collection()
print(f"\n📊 Total KPIs across all scenarios: {all_kpis.size}")

# Clear existing KPIs study.scen.clear_kpi_collection_for_all_child_datasets() # Generate price KPIs for all scenarios price_def = ( kpis.FlagAggKPIBuilder() .for_flag('control_areas_t.vol_weighted_marginal_price') .with_aggregation(kpis.Aggregations.Mean) .for_all_objects() .build() ) for dataset in study.scen.datasets: dataset.add_kpis_from_definitions(price_def) print(f"✅ Generated {len(price_def)} KPIs for {dataset.name}") # Get merged collection all_kpis = study.scen.get_merged_kpi_collection() print(f"\n📊 Total KPIs across all scenarios: {all_kpis.size}")

✅ Generated 1 KPIs for base
✅ Generated 1 KPIs for solar_150
✅ Generated 1 KPIs for solar_200
✅ Generated 1 KPIs for wind_150
✅ Generated 1 KPIs for wind_200

📊 Total KPIs across all scenarios: 20

Part 6: KPI Collection Filtering¶

The KPI collection provides powerful filtering capabilities:

In [12]:

# Filter by attributes
tennet_kpis = all_kpis.filter(object_name='TenneTDE')
print(f"KPIs for TenneTDE: {tennet_kpis.size}")

# Filter by dataset
base_kpis = all_kpis.filter(dataset_name='base')
print(f"KPIs for base scenario: {base_kpis.size}")

# Combined filtering
tennet_base = all_kpis.filter(
    object_name='TenneTDE',
    dataset_name='base',
    aggregation=kpis.Aggregations.Mean
)
print(f"\nFiltered result: {tennet_base.size} KPI(s)")
if tennet_base.size > 0:
    kpi = tennet_base[0]
    print(f"  {kpi.name}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

# Filter by attributes tennet_kpis = all_kpis.filter(object_name='TenneTDE') print(f"KPIs for TenneTDE: {tennet_kpis.size}") # Filter by dataset base_kpis = all_kpis.filter(dataset_name='base') print(f"KPIs for base scenario: {base_kpis.size}") # Combined filtering tennet_base = all_kpis.filter( object_name='TenneTDE', dataset_name='base', aggregation=kpis.Aggregations.Mean ) print(f"\nFiltered result: {tennet_base.size} KPI(s)") if tennet_base.size > 0: kpi = tennet_base[0] print(f" {kpi.name}: {Units.get_pretty_text_for_quantity(kpi.quantity)}")

KPIs for TenneTDE: 5
KPIs for base scenario: 4

Filtered result: 1 KPI(s)
  control_areas_t.vol_weighted_marginal_price Mean TenneTDE: 16.9 €/MWh

Part 7: KPI Grouping¶

Group KPIs by attributes for analysis:

In [13]:

# Group by dataset
by_dataset = all_kpis.group_by('dataset_name')

print("KPIs grouped by dataset:")
for (dataset,), kpi_collection in by_dataset.items():
    print(f"  {dataset}: {kpi_collection.size} KPIs")

# Group by object and dataset
by_object_dataset = all_kpis.group_by('object_name', 'dataset_name')

print(f"\nTotal groups (object × dataset): {len(by_object_dataset)}")
print("\nSample groups:")
for i, ((obj, ds), kpis) in enumerate(by_object_dataset.items()):
    if i < 3:
        print(f"  {obj} × {ds}: {kpis.size} KPI(s)")

# Group by dataset by_dataset = all_kpis.group_by('dataset_name') print("KPIs grouped by dataset:") for (dataset,), kpi_collection in by_dataset.items(): print(f" {dataset}: {kpi_collection.size} KPIs") # Group by object and dataset by_object_dataset = all_kpis.group_by('object_name', 'dataset_name') print(f"\nTotal groups (object × dataset): {len(by_object_dataset)}") print("\nSample groups:") for i, ((obj, ds), kpis) in enumerate(by_object_dataset.items()): if i < 3: print(f" {obj} × {ds}: {kpis.size} KPI(s)")

KPIs grouped by dataset:
  base: 4 KPIs
  solar_150: 4 KPIs
  solar_200: 4 KPIs
  wind_150: 4 KPIs
  wind_200: 4 KPIs

Total groups (object × dataset): 20

Sample groups:
  50Hertz × base: 1 KPI(s)
  Amprion × base: 1 KPI(s)
  TenneTDE × base: 1 KPI(s)

Summary: KPI Framework Capabilities¶

Key Features Demonstrated:¶

1. Fluent Builder API

   • .for_flag() - Specify data source
   • .with_aggregation() - Choose aggregation type
   • .for_objects() - Select objects
   • .with_attributes() - Add custom metadata
   • .build() - Create definition

2. Batch Operations

   • Single fetch per flag (not per object)
   • Column-wise aggregation
   • 10x performance improvement

3. Unit Handling

   • Automatic unit tracking
   • Pretty unit conversion
   • get_kpi_in_unit() - Convert to specific unit
   • get_kpi_in_pretty_unit() - Auto-select readable unit

4. Rich Metadata

   • Object name, flag, aggregation
   • Dataset name and type
   • Custom attributes
   • Extra attributes dictionary

5. Collection Operations

   • .filter() - Filter by attributes
   • .group_by() - Group by attributes
   • .get_related() - Find related KPIs
   • .to_dataframe() - Export to pandas

Performance Benefits:¶

• Old System: N fetches + N kpis.Aggregations (one per object)
• New System: 1 fetch + 1 aggregation (batch operation)
• Result: Up to 10x faster for large datasets

Next Steps:¶

In notebook 202, we'll explore:

• Exporting KPI collections to DataFrames
• Creating pretty tables with automatic unit normalization
• Comparison KPIs for scenario analysis
• Advanced filtering and visualization

Conclusion¶

The MESQUAL KPI framework provides:

• ✅ Clean, declarative API - Builder pattern for readable code
• ✅ High performance - Batch operations for efficiency
• ✅ Automatic unit handling - No manual conversion needed
• ✅ Rich metadata - Everything you need for filtering and analysis
• ✅ Flexible filtering - Find exactly the KPIs you need s This foundation enables sophisticated multi-scenario energy system analysis with minimal code.