Bases: KPIDefinition
Simple flag + aggregation KPI definition.
Computes KPIs by:
1. Fetching flag data (DataFrame with objects as columns)
2. Applying aggregation (vectorized across all columns)
3. Creating one KPI per object (column)
This is the most common KPI type for energy system analysis.
Source code in submodules/mesqual/mesqual/kpis/definitions/flag_aggregation.py
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 | @dataclass
class FlagAggKPIDefinition(KPIDefinition):
"""
Simple flag + aggregation KPI definition.
Computes KPIs by:
1. Fetching flag data (DataFrame with objects as columns)
2. Applying aggregation (vectorized across all columns)
3. Creating one KPI per object (column)
This is the most common KPI type for energy system analysis.
"""
flag: FlagTypeProtocol
aggregation: Aggregation
model_flag: FlagTypeProtocol | None = None # Auto-inferred if None
objects: list[Hashable] | Literal['auto'] | ModelPropertyFilter = 'auto' # 'auto' = discover from data
extra_attributes: dict = None
name_prefix: str = ''
name_suffix: str = ''
custom_name: str | None = None # Complete name override
def generate_kpis(self, dataset: Dataset) -> list[KPI]:
"""
Generate KPIs by batch computation.
Process:
1. Infer model_flag if not provided
2. Fetch data (DataFrame)
3. Discover objects from DataFrame columns if objects='auto'
4. Apply aggregation to entire DataFrame (vectorized)
5. Create KPI instance per object with metadata
Args:
dataset: Dataset to compute KPIs for
Returns:
List of computed KPI instances
"""
from mesqual.kpis.builders.flag_agg_builder import ModelPropertyFilter
model_flag = self.model_flag or dataset.flag_index.get_linked_model_flag(self.flag)
df = dataset.fetch(self.flag)
model_df = dataset.fetch(model_flag) if dataset.flag_is_accepted(model_flag) else None
if self.objects == 'auto':
objects = df.columns.tolist()
elif isinstance(self.objects, ModelPropertyFilter):
if model_df is None:
raise Exception(f'No Model DF found for model_flag "{model_flag}" in variable_flag "{self.flag}"')
filtered_model_df_objects = self.objects.apply_filter(model_df)
objects = [o for o in df.columns if o in filtered_model_df_objects]
else:
objects = self.objects
aggregated = self.aggregation(df) # Returns Series with one value per column
dataset_type = type(dataset)
unit = dataset.flag_index.get_unit(self.flag)
if self.aggregation.unit is not None:
unit = self.aggregation.unit
kpis = []
for obj in objects:
if obj not in aggregated.index:
# TODO: optional warning if object listed but not present in flag
continue # Skip objects not in aggregated results
# If custom_name is set and there are multiple objects, append object name
kpi_custom_name = self.custom_name
if self.custom_name and len(objects) > 1:
kpi_custom_name = f"{self.custom_name} {obj}"
attributes = KPIAttributes(
flag=self.flag,
model_flag=model_flag,
object_name=obj,
aggregation=self.aggregation,
dataset_name=dataset.name,
dataset_type=dataset_type,
name_prefix=self.name_prefix,
name_suffix=self.name_suffix,
custom_name=kpi_custom_name,
unit=unit,
dataset_attributes=dataset.attributes,
extra_attributes=self.extra_attributes or dict()
)
kpi = KPI(
value=aggregated[obj],
attributes=attributes,
dataset=dataset
)
if (model_df is not None) and (obj in model_df.index):
kpi._object_info = model_df.loc[obj]
kpis.append(kpi)
return kpis
def required_flags(self) -> set[FlagTypeProtocol]:
"""
Return required flags.
Returns:
Set of required flags
"""
flags = {self.flag}
if self.model_flag:
flags.add(self.model_flag)
return flags
|
Generate KPIs by batch computation.
Process:
1. Infer model_flag if not provided
2. Fetch data (DataFrame)
3. Discover objects from DataFrame columns if objects='auto'
4. Apply aggregation to entire DataFrame (vectorized)
5. Create KPI instance per object with metadata
Parameters:
| Name |
Type |
Description |
Default |
dataset
|
Dataset
|
Dataset to compute KPIs for
|
required
|
Returns:
| Type |
Description |
list[KPI]
|
List of computed KPI instances
|
Source code in submodules/mesqual/mesqual/kpis/definitions/flag_aggregation.py
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 | def generate_kpis(self, dataset: Dataset) -> list[KPI]:
"""
Generate KPIs by batch computation.
Process:
1. Infer model_flag if not provided
2. Fetch data (DataFrame)
3. Discover objects from DataFrame columns if objects='auto'
4. Apply aggregation to entire DataFrame (vectorized)
5. Create KPI instance per object with metadata
Args:
dataset: Dataset to compute KPIs for
Returns:
List of computed KPI instances
"""
from mesqual.kpis.builders.flag_agg_builder import ModelPropertyFilter
model_flag = self.model_flag or dataset.flag_index.get_linked_model_flag(self.flag)
df = dataset.fetch(self.flag)
model_df = dataset.fetch(model_flag) if dataset.flag_is_accepted(model_flag) else None
if self.objects == 'auto':
objects = df.columns.tolist()
elif isinstance(self.objects, ModelPropertyFilter):
if model_df is None:
raise Exception(f'No Model DF found for model_flag "{model_flag}" in variable_flag "{self.flag}"')
filtered_model_df_objects = self.objects.apply_filter(model_df)
objects = [o for o in df.columns if o in filtered_model_df_objects]
else:
objects = self.objects
aggregated = self.aggregation(df) # Returns Series with one value per column
dataset_type = type(dataset)
unit = dataset.flag_index.get_unit(self.flag)
if self.aggregation.unit is not None:
unit = self.aggregation.unit
kpis = []
for obj in objects:
if obj not in aggregated.index:
# TODO: optional warning if object listed but not present in flag
continue # Skip objects not in aggregated results
# If custom_name is set and there are multiple objects, append object name
kpi_custom_name = self.custom_name
if self.custom_name and len(objects) > 1:
kpi_custom_name = f"{self.custom_name} {obj}"
attributes = KPIAttributes(
flag=self.flag,
model_flag=model_flag,
object_name=obj,
aggregation=self.aggregation,
dataset_name=dataset.name,
dataset_type=dataset_type,
name_prefix=self.name_prefix,
name_suffix=self.name_suffix,
custom_name=kpi_custom_name,
unit=unit,
dataset_attributes=dataset.attributes,
extra_attributes=self.extra_attributes or dict()
)
kpi = KPI(
value=aggregated[obj],
attributes=attributes,
dataset=dataset
)
if (model_df is not None) and (obj in model_df.index):
kpi._object_info = model_df.loc[obj]
kpis.append(kpi)
return kpis
|
required_flags() -> set[FlagTypeProtocol]
Return required flags.
Returns:
| Type |
Description |
set[FlagTypeProtocol]
|
|
Source code in submodules/mesqual/mesqual/kpis/definitions/flag_aggregation.py
120
121
122
123
124
125
126
127
128
129
130 | def required_flags(self) -> set[FlagTypeProtocol]:
"""
Return required flags.
Returns:
Set of required flags
"""
flags = {self.flag}
if self.model_flag:
flags.add(self.model_flag)
return flags
|