Coverage for src/sparkle/structures/feature_dataframe.py: 95%
146 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1"""Module to manage feature data files and common operations on them."""
3from __future__ import annotations
4import math
5from pathlib import Path
7import pandas as pd
10class FeatureDataFrame(pd.DataFrame):
11 """Class to manage feature data CSV files and common operations on them."""
13 missing_value = math.nan
14 extractor_dim = "Extractor"
15 feature_group_dim = "FeatureGroup"
16 feature_name_dim = "FeatureName"
17 instance_set_index_dim = "InstanceSet"
18 instance_index_dim = "Instance"
19 multi_dim_index_names = [instance_set_index_dim, instance_index_dim]
20 multi_dim_column_names = [extractor_dim, feature_group_dim, feature_name_dim]
22 def __init__(
23 self: FeatureDataFrame,
24 csv_filepath: Path,
25 instance_pairs: list[tuple[str, str]] = [],
26 extractor_data: dict[str, list[tuple[str, str]]] = {},
27 ) -> None:
28 """Initialise a FeatureDataFrame object.
30 Arguments:
31 csv_filepath: The Path for the CSV storage. If it does not exist,
32 a new DataFrame will be initialised and stored here.
33 instance_pairs: The list of (set_name, instance_name) pairs to be added.
34 extractor_data: A dictionary with extractor names as key, and a list of
35 tuples ordered as [(feature_group, feature_name), ...] as value.
36 """
37 # Initialize a dataframe from an existing file
38 if csv_filepath.exists():
39 # Read the 2-level (InstanceSet, Instance) row index and 3-level
40 # (Extractor, FeatureGroup, FeatureName) column header.
41 temp_df = pd.read_csv(
42 csv_filepath,
43 header=[0, 1, 2],
44 index_col=[0, 1],
45 dtype={
46 FeatureDataFrame.extractor_dim: str,
47 FeatureDataFrame.feature_group_dim: str,
48 FeatureDataFrame.feature_name_dim: str,
49 },
50 on_bad_lines="skip",
51 skip_blank_lines=True,
52 )
53 temp_df.index.names = FeatureDataFrame.multi_dim_index_names
54 super().__init__(temp_df)
55 self.csv_filepath = csv_filepath
56 # Create a new dataframe
57 else:
58 # Unfold the extractor_data into lists
59 if extractor_data:
60 multi_column_lists = [
61 (extractor, group, feature_name)
62 for extractor in extractor_data
63 for group, feature_name in extractor_data[extractor]
64 ]
65 else:
66 multi_column_lists = [
67 (
68 FeatureDataFrame.missing_value,
69 FeatureDataFrame.missing_value,
70 FeatureDataFrame.feature_name_dim,
71 )
72 ]
73 # Initialise new dataframe
74 multi_columns = pd.MultiIndex.from_tuples(
75 multi_column_lists, names=self.multi_dim_column_names
76 )
77 if instance_pairs:
78 index = pd.MultiIndex.from_tuples(
79 instance_pairs, names=self.multi_dim_index_names
80 )
81 else:
82 index = pd.MultiIndex.from_tuples([], names=self.multi_dim_index_names)
83 super().__init__(
84 data=self.missing_value,
85 index=index,
86 columns=multi_columns,
87 dtype=float,
88 )
89 self.csv_filepath = csv_filepath
90 self.save_csv()
92 if self.index.duplicated().any(): # Drop all duplicates except for last
93 self.reset_index(inplace=True) # Reset index to columns
94 idx_cols = self.columns[:2].tolist() # Both InstanceSet and Instance cols
95 self.drop_duplicates(
96 subset=idx_cols, keep="last", inplace=True
97 ) # filter duplicates from index columns
98 self.set_index(idx_cols, inplace=True) # Restore the MultiIndex (in-place)
99 self.index.names = FeatureDataFrame.multi_dim_index_names
101 # Sort the index to optimize lookup speed
102 self.sort_index(axis=0, inplace=True)
103 self.sort_index(axis=1, inplace=True)
105 def add_extractor(
106 self: FeatureDataFrame,
107 extractor: str,
108 extractor_features: list[tuple[str, str]],
109 values: list[list[float]] = None,
110 ) -> None:
111 """Add an extractor and its feature names to the dataframe.
113 Arguments:
114 extractor: Name of the extractor
115 extractor_features: Tuples of [FeatureGroup, FeatureName]
116 values: Initial values of the Extractor per instance in the dataframe.
117 Defaults to FeatureDataFrame.missing_value.
118 """
119 if extractor in self.extractors:
120 print(
121 f"WARNING: Tried adding already existing extractor {extractor} to "
122 f"Feature DataFrame: {self.csv_filepath}"
123 )
124 return
125 if values is None:
126 values = [self.missing_value] * len(
127 extractor_features
128 ) # Single missing value for each feature
129 extractor_dim = self.columns.get_level_values(FeatureDataFrame.extractor_dim)
130 # Unfold to indices to lists
131 for index, (feature_group, feature) in enumerate(extractor_features):
132 self[(extractor, feature_group, feature)] = values[index]
133 if self.num_extractors > 1:
134 # Upon successfull adding of the extractor, remove the nan extractor
135 if str(math.nan) in extractor_dim:
136 self.drop(
137 str(math.nan),
138 axis=1,
139 level=FeatureDataFrame.extractor_dim,
140 inplace=True,
141 )
142 elif math.nan in extractor_dim:
143 self.drop(
144 math.nan, axis=1, level=FeatureDataFrame.extractor_dim, inplace=True
145 )
147 def add_instance(
148 self: FeatureDataFrame,
149 instance_pairs: tuple[str, str] | list[tuple[str, str]],
150 values: list[float] = None,
151 ) -> None:
152 """Add one or more instances to the dataframe.
154 Args:
155 instance_pairs: A (set_name, instance_name) pair or list of such pairs.
156 values: Optional initial values for all features. The same values are
157 used for every added instance.
158 """
159 if isinstance(instance_pairs, tuple):
160 instance_pairs = [instance_pairs]
161 fill = values if values else FeatureDataFrame.missing_value
162 row_values = fill if isinstance(fill, list) else [fill] * len(self.columns)
163 for instance_pair in instance_pairs:
164 self.loc[instance_pair, :] = row_values
166 def remove_extractor(self: FeatureDataFrame, extractor: str) -> None:
167 """Remove an extractor from the dataframe."""
168 self.drop(extractor, axis=1, level=FeatureDataFrame.extractor_dim, inplace=True)
169 # if self.num_extractors == 0:
170 if self.num_extractors == 0: # make sure we have atleast one 'extractor'
171 self.add_extractor(
172 str(FeatureDataFrame.missing_value),
173 [(FeatureDataFrame.missing_value, FeatureDataFrame.feature_name_dim)],
174 )
176 def remove_instance(
177 self: FeatureDataFrame,
178 instance_pairs: tuple[str, str] | list[tuple[str, str]],
179 ) -> None:
180 """Remove one or more instances from the dataframe.
182 Args:
183 instance_pairs: A (set_name, instance_name) pair or list of such pairs.
184 """
185 if isinstance(instance_pairs, tuple):
186 instance_pairs = [instance_pairs]
187 self.drop(instance_pairs, axis=0, inplace=True)
189 def get_feature_groups(
190 self: FeatureDataFrame, extractor: str | list[str] = None
191 ) -> list[str]:
192 """Retrieve the feature groups in the dataframe.
194 Args:
195 extractor: Optional. If extractor(s) are given,
196 yields only feature groups of that extractor.
198 Returns:
199 A list of feature groups.
200 """
201 columns = self.columns
202 if extractor is not None:
203 if isinstance(extractor, str):
204 extractor = [extractor]
205 columns = columns[columns.isin(extractor, level=0)]
206 return columns.get_level_values(level=1).unique().to_list()
208 def get_value(
209 self: FeatureDataFrame,
210 instance_set: str,
211 instance: str,
212 extractor: str,
213 feature_group: str,
214 feature_name: str,
215 ) -> float:
216 """Return a value in the dataframe.
218 Args:
219 instance_set: Name of the instance set the instance belongs to.
220 instance: Name of the instance.
221 extractor: Name of the extractor.
222 feature_group: Name of the feature group.
223 feature_name: Name of the feature.
225 Returns:
226 The value.
227 """
228 return self.loc[
229 (instance_set, instance), (extractor, feature_group, feature_name)
230 ]
232 def set_value(
233 self: FeatureDataFrame,
234 instance_set: str,
235 instance: str,
236 extractor: str,
237 feature_group: str,
238 feature_name: str | list[str],
239 value: float | list[float],
240 append_write_csv: bool = False,
241 ) -> None:
242 """Set a value in the dataframe.
244 Args:
245 instance_set: Name of the instance set the instance belongs to.
246 instance: Name of the instance.
247 extractor: Name of the extractor.
248 feature_group: Name of the feature group.
249 feature_name: Name of the feature.
250 value: The value to set.
251 append_write_csv: CSV to be written to.
252 """
253 if isinstance(feature_name, list) and isinstance(value, list):
254 if len(feature_name) != len(value):
255 raise ValueError(
256 f"feature_name and values must be the same length ({len(feature_name)}, {len(value)})."
257 )
258 elif isinstance(feature_name, list) or isinstance(value, list):
259 raise ValueError(
260 f"feature_name parameter and value must be the same type ({type(feature_name)}, {type(value)})."
261 )
262 instance_pair = (instance_set, instance)
263 self.loc[instance_pair, (extractor, feature_group, feature_name)] = value
264 if append_write_csv:
265 writeable = self.loc[[instance_pair], :] # Take line
266 # Append the new rows to the dataframe csv file
267 import os
269 csv_string = writeable.to_csv(header=False) # Convert to the csv lines
270 for line in csv_string.splitlines(): # Should be only one line, but is safe now if we were to do multiple values
271 fd = os.open(f"{self.csv_filepath}", os.O_WRONLY | os.O_APPEND)
272 os.write(fd, f"{line}\n".encode("utf-8")) # Encode to create buffer
273 # Open and close for each line to minimise possibilities of conflict
274 os.close(fd)
276 def has_missing_vectors(self: FeatureDataFrame) -> bool:
277 """Returns True if there are any Extractors still to be run on any instance."""
278 for extractor in self.extractors:
279 # True if any instance has ALL features null for this extractor
280 if self[extractor].isnull().all(axis=1).any():
281 return True
282 return False
284 def remaining_jobs(
285 self: FeatureDataFrame,
286 groupwise_computation: bool = True,
287 ) -> list[tuple[tuple[str, str], str, str | None]]:
288 """Return remaining feature-computation jobs.
290 Args:
291 groupwise_computation:
292 If True, jobs are kept per feature group and returned as
293 `((set_name, instance_name), extractor_name, feature_group)` tuples.
294 If False, feature groups are collapsed and the return value uses
295 `None` for the feature-group position:
296 `((set_name, instance_name), extractor_name, None)`.
298 Returns:
299 A flat list of remaining jobs, always in the shape
300 `((set_name, instance_name), extractor_name, feature_group | None)`.
301 """
302 extractor_values = self.extractors
304 # DataFrame restricted to real extractor columns only.
305 target_df = self.loc[:, extractor_values]
307 if target_df.empty:
308 return []
310 # Build one boolean per (instance, extractor, feature_group):
311 # 1) target_df.isnull(): mark missing cells as True.
312 # 2) .T: move feature columns to the index for grouping by MultiIndex levels.
313 # 3) .groupby(level=[Extractor, FeatureGroup]).all():
314 # collapse all feature names in the same group.
315 # Result is True only if the *entire* group is missing for an instance.
316 # 4) final .T: restore instances on rows.
317 # So missing_groups.loc[instance, (extractor, feature_group)] == True
318 # means this job still has to be computed.
319 missing_groups = (
320 target_df.isnull()
321 .T.groupby(
322 level=[
323 FeatureDataFrame.extractor_dim,
324 FeatureDataFrame.feature_group_dim,
325 ]
326 )
327 .all()
328 .T
329 )
330 if groupwise_computation:
331 # Convert the 2D table to a Series with MultiIndex:
332 # (set_name, instance_name, extractor, feature_group) -> bool.
333 stacked_missing = missing_groups.stack(
334 [FeatureDataFrame.extractor_dim, FeatureDataFrame.feature_group_dim],
335 future_stack=True,
336 )
337 # Keep only True entries and repack as ((set_name, instance_name), extractor, feature_group).
338 return [
339 ((set_name, instance_name), extractor, feature_group)
340 for set_name, instance_name, extractor, feature_group in stacked_missing[
341 stacked_missing
342 ].index.to_list()
343 ]
345 # Collapse feature groups into one boolean per (instance, extractor):
346 missing_values_with_no_group = (
347 missing_groups.T.groupby(level=[FeatureDataFrame.extractor_dim]).all().T
348 )
349 # Convert collapsed table to Series:
350 # (set_name, instance_name, extractor) -> bool.
351 stacked_missing = missing_values_with_no_group.stack(
352 [FeatureDataFrame.extractor_dim],
353 future_stack=True,
354 )
356 # Keep only True entries and expand to 3-tuple with feature-group slot = None.
357 return [
358 ((set_name, instance_name), extractor, None)
359 for set_name, instance_name, extractor in stacked_missing[
360 stacked_missing
361 ].index.to_list()
362 ]
364 def get_instance(
365 self: FeatureDataFrame,
366 instance_set: str,
367 instance: str,
368 as_dataframe: bool = False,
369 ) -> list[float]:
370 """Return the feature vector of an instance pair.
372 Args:
373 instance_set: Name of the instance set the instance belongs to.
374 instance: Name of the instance.
375 as_dataframe: True if instances should be returned as df.
377 Returns:
378 The feature vector of an instance pair.
379 """
380 instance_pair = (instance_set, instance)
381 if as_dataframe:
382 return self.loc[[instance_pair]]
383 return self.loc[instance_pair].tolist()
385 def impute_missing_values(self: FeatureDataFrame) -> None:
386 """Imputes all NaN values by taking the average feature value."""
387 # imputed_df = self.T.fillna(self.mean(axis=1)).T
388 imputed_df = self.fillna(self.mean(axis=0))
389 self[:] = imputed_df.values
391 def has_missing_value(self: FeatureDataFrame) -> bool:
392 """Return whether there are missing values in the feature data."""
393 return self.isnull().any().any()
395 def reset_dataframe(self: FeatureDataFrame) -> bool:
396 """Resets all values to FeatureDataFrame.missing_value."""
397 self.loc[:, (slice(None), slice(None), slice(None))] = (
398 FeatureDataFrame.missing_value
399 )
401 def sort(self: FeatureDataFrame) -> None:
402 """Sorts the DataFrame by Multi-Index for readability."""
403 self.sort_index(inplace=True)
405 @property
406 def instance_pairs(self: FeatureDataFrame) -> list[tuple[str, str]]:
407 """Return the (set_name, instance_name) pairs in the dataframe."""
408 return self.index.tolist()
410 @property
411 def extractors(self: FeatureDataFrame) -> list[str]:
412 """Returns all unique extractors in the DataFrame."""
413 return [
414 x
415 for x in self.columns.get_level_values(
416 FeatureDataFrame.extractor_dim
417 ).unique()
418 if str(x) != str(FeatureDataFrame.missing_value)
419 ]
421 @property
422 def num_features(self: FeatureDataFrame) -> int:
423 """Return the number of features in the dataframe."""
424 # return self.shape[0]
425 return self.shape[1]
427 @property
428 def num_instances(self: FeatureDataFrame) -> int:
429 """Return the number of instances in the dataframe."""
430 # return self.shape[1]
431 return self.shape[0]
433 @property
434 def num_extractors(self: FeatureDataFrame) -> int:
435 """Return the number of extractors in the dataframe."""
436 return self.columns.get_level_values("Extractor").unique().size
438 @property
439 def features(self: FeatureDataFrame) -> list[str]:
440 """Return the features in the dataframe."""
441 # return self.index.get_level_values("FeatureName").unique().to_list()
442 return self.columns.get_level_values("FeatureName").unique().to_list()
444 def save_csv(self: FeatureDataFrame, csv_filepath: Path = None) -> None:
445 """Write a CSV to the given path.
447 Args:
448 csv_filepath: String path to the csv file. Defaults to self.csv_filepath.
449 """
450 csv_filepath = self.csv_filepath if csv_filepath is None else csv_filepath
451 if csv_filepath is None:
452 raise ValueError("Cannot save DataFrame: no `csv_filepath` was provided.")
453 self.sort_index(inplace=True)
454 self.to_csv(csv_filepath)