From 5463d09167ad19ed355310b0319ed3dfbcbedc68 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 23 Nov 2025 09:38:43 -0500 Subject: [PATCH 01/12] Improve hdf5Reader with random sampling and type handling Enhanced the hdf5Reader class to include random sampling of rows while reading HDF5 files. Improved handling of numpy types and added error handling for various data processing steps. --- aidrin/file_handling/readers/hdf5_reader.py | 213 ++++++++++++-------- 1 file changed, 126 insertions(+), 87 deletions(-) diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py index 2d56a6e2..54be7635 100644 --- a/aidrin/file_handling/readers/hdf5_reader.py +++ b/aidrin/file_handling/readers/hdf5_reader.py @@ -1,8 +1,9 @@ import os import uuid - +import random import h5py import pandas as pd +import numpy as np from flask import current_app, session from aidrin.file_handling.readers.base_reader import BaseFileReader @@ -11,138 +12,177 @@ class hdf5Reader(BaseFileReader): def read(self): try: + # Limit number of rows using random sampling + MAX_ROWS = 2000 rows = [] - # Clean up byte strings in all object columns + count = 0 + + # Clean up byte strings in all object columns def decode_bytes(df): for col in df.columns: if df[col].dtype == object: df[col] = df[col].apply( - lambda x: x.decode("utf-8") if isinstance(x, bytes) else x + lambda x: x.decode("utf-8") if isinstance(x, (bytes, bytearray, np.bytes_)) else x ) return df + # Expand structured numpy dtypes into list of dicts + def structured_to_records(data): + try: + if isinstance(data, np.ndarray): + if getattr(data, "dtype", None) is not None and data.dtype.names: + return [dict(zip(data.dtype.names, row)) for row in data] + except Exception: + pass + return data + + # Convert numpy types to Python native types def convert_numpy_types(obj): - """Recursively convert numpy types to Python native types""" try: - if hasattr(obj, 'item'): # numpy scalar + if hasattr(obj, "item"): return obj.item() elif isinstance(obj, (list, tuple)): - return [convert_numpy_types(item) for item in obj] + return [convert_numpy_types(x) for x in obj] elif isinstance(obj, dict): return {str(k): convert_numpy_types(v) for k, v in obj.items()} - elif hasattr(obj, 'dtype'): # numpy array - if obj.size == 1: # Single element array + elif isinstance(obj, np.ndarray): + if obj.size == 1: return obj.item() - else: # Multi-element array - return obj.tolist() - else: - return obj - except Exception as e: - self.logger.warning(f"Error converting numpy type: {e}") - return str(obj) # Fallback to string representation + return [convert_numpy_types(x) for x in obj.tolist()] + return obj + except Exception: + return str(obj) + + # Random sampling helper + def add_row(row_dict): + nonlocal count + count += 1 + if len(rows) < MAX_ROWS: + rows.append(row_dict) + else: + j = random.randint(1, count) + if j <= MAX_ROWS: + rows[j - 1] = row_dict + + # Process a chunk of data (structured dtype, ND flattening, sampling) + def process_data_chunk(data): + data = structured_to_records(data) + + # ND flattening for multidimensional arrays + if isinstance(data, np.ndarray) and hasattr(data, "ndim") and data.ndim > 2: + try: + data = np.ascontiguousarray(data).reshape(data.shape[0], -1) + except Exception: + return + + # Fast-path for simple 1D arrays + if isinstance(data, np.ndarray) and data.ndim == 1: + for val in data: + try: + row_val = convert_numpy_types(val) + add_row({"value": row_val}) + except Exception: + continue + return - def recurse(name, obj, path=[]): - try: - if isinstance(obj, h5py.Dataset): - data = obj[()] - # If it's a 1D or structured dataset, load it into dicts - if isinstance(data, (list, tuple)) or hasattr(data, "dtype"): + # Handle table-like datasets + if isinstance(data, (list, tuple)) or hasattr(data, "dtype"): + try: + df = pd.DataFrame(data) + except Exception: + try: + df = pd.DataFrame(data.tolist()) + except Exception: + return + + for _, row in df.iterrows(): + try: + row_dict = convert_numpy_types(row.to_dict()) + add_row(row_dict) + except Exception: try: - df = pd.DataFrame(data) + basic_row = {} + for col in row.index: + val = row[col] + if hasattr(val, "item"): + basic_row[str(col)] = val.item() + else: + basic_row[str(col)] = str(val) + add_row(basic_row) except Exception: - df = pd.DataFrame(data.tolist()) # base - for _, row in df.iterrows(): - try: - row_dict = row.to_dict() - # Convert any numpy types to Python native types - row_dict = convert_numpy_types(row_dict) - rows.append(row_dict) - except Exception as e: - self.logger.warning(f"Error processing row: {e}") - # Try to process the row with basic conversion - try: - basic_row = {} - for col in row.index: - try: - value = row[col] - if hasattr(value, 'item'): - basic_row[str(col)] = value.item() - else: - basic_row[str(col)] = str(value) - except Exception: - basic_row[str(col)] = str(value) - rows.append(basic_row) - except Exception as e2: - self.logger.warning(f"Failed to process row even with basic conversion: {e2}") - continue - else: - # Scalar or flat dataset - ensure data is hashable + continue + else: + try: + val = convert_numpy_types(data) + add_row({"value": val}) + except Exception: + try: + if hasattr(data, "item"): + add_row({"value": data.item()}) + else: + add_row({"value": str(data)}) + except Exception: + return + + # Chunked dataset reading + sampling + def recurse(name, obj, path=[]): + if isinstance(obj, h5py.Dataset): + shape = getattr(obj, "shape", None) + + # Chunk mode for any large dataset (Option 2) + if shape is not None and len(shape) > 0 and shape[0] > 10000: + step = 10000 + for i in range(0, shape[0], step): try: - # Convert any numpy types to Python native types - data = convert_numpy_types(data) - row_dict = {"value": data} - rows.append(row_dict) - except Exception as e: - self.logger.warning(f"Error processing scalar data: {e}") - # Try basic conversion - try: - if hasattr(data, 'item'): - row_dict = {"value": data.item()} - else: - row_dict = {"value": str(data)} - rows.append(row_dict) - except Exception as e2: - self.logger.warning(f"Failed to process scalar data even with basic conversion: {e2}") - # Skip this data point - pass - except Exception as e: - self.logger.warning(f"Error in recurse function: {e}") - return + chunk = obj[i:i+step] + process_data_chunk(chunk) + except Exception: + break + else: + try: + data = obj[()] + process_data_chunk(data) + except Exception: + return + # Traverse all datasets with h5py.File(self.file_path, "r") as f: - def visit(name, obj): recurse(name, obj, name.strip("/").split("/")) - f.visititems(visit) + + # Create final DataFrame df = pd.DataFrame(rows) df = decode_bytes(df) - # Ensure all column names are strings to avoid numpy array issues - if hasattr(df, 'columns') and len(df.columns) > 0: + if hasattr(df, "columns") and len(df.columns) > 0: df.columns = [str(col) for col in df.columns] - # Check if DataFrame is empty and log warning if df.empty: self.logger.warning("No data was successfully processed from HDF5 file") return None return df + except Exception as e: self.logger.error(f"Error while reading: {e}") return None def parse(self): - # Recursively find all group names in the HDF5 file def recurse(data): try: - # Convert items() to a list first to avoid iteration issues items = list(data.items()) for name, obj in items: try: - # Ensure name is a string and hashable to avoid "unhashable type" errors full_path = str(name) - if isinstance(obj, h5py.Group): group_names.append(full_path) recurse(obj) - except (TypeError, ValueError) as e: - # If conversion fails, skip this key and log the error - self.logger.warning(f"Skipping unhashable key {name}: {e}") + except (TypeError, ValueError): continue - except Exception as e: - self.logger.error(f"Error during recursion: {e}") + except Exception: + pass return group_names with h5py.File(self.file_path, "r") as f: @@ -154,16 +194,13 @@ def recurse(data): def filter(self, kept_keys): if isinstance(kept_keys, str): kept_keys = kept_keys.split(",") - # Ensure all keys are strings and hashable to avoid "unhashable type" errors + filtered_keys = set() for g in kept_keys: try: - # Convert to string and ensure it's hashable key_str = str(g).strip("/") - # Test if it's hashable by trying to add to set filtered_keys.add(key_str) except (TypeError, ValueError) as e: - # If conversion fails, skip this key and log the error self.logger.warning(f"Skipping unhashable key {g}: {e}") continue @@ -171,6 +208,7 @@ def filter(self, kept_keys): f"filtered_{uuid.uuid4().hex}_{session.get('uploaded_file_name')}" ) new_file_path = os.path.join(current_app.config["UPLOAD_FOLDER"], new_file_name) + with ( h5py.File(self.file_path, "r") as src, h5py.File(new_file_path, "w") as tgt, @@ -192,3 +230,4 @@ def copy_group(path, src_group, tgt_group): copy_group("", src, tgt) return new_file_path + From 28f79f500a90b12a2c9c4ec8afacafc129d1e795 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 30 Nov 2025 11:15:26 -0500 Subject: [PATCH 02/12] Update hdf5_reader.py --- aidrin/file_handling/readers/hdf5_reader.py | 41 +++++++-------------- 1 file changed, 14 insertions(+), 27 deletions(-) diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py index 54be7635..a25cf832 100644 --- a/aidrin/file_handling/readers/hdf5_reader.py +++ b/aidrin/file_handling/readers/hdf5_reader.py @@ -12,18 +12,16 @@ class hdf5Reader(BaseFileReader): def read(self): try: - # Limit number of rows using random sampling - MAX_ROWS = 2000 - rows = [] - count = 0 - + rows = [] # Clean up byte strings in all object columns def decode_bytes(df): for col in df.columns: if df[col].dtype == object: df[col] = df[col].apply( - lambda x: x.decode("utf-8") if isinstance(x, (bytes, bytearray, np.bytes_)) else x + lambda x: x.decode("utf-8") + if isinstance(x, (bytes, bytearray, np.bytes_)) + else x ) return df @@ -54,18 +52,7 @@ def convert_numpy_types(obj): except Exception: return str(obj) - # Random sampling helper - def add_row(row_dict): - nonlocal count - count += 1 - if len(rows) < MAX_ROWS: - rows.append(row_dict) - else: - j = random.randint(1, count) - if j <= MAX_ROWS: - rows[j - 1] = row_dict - - # Process a chunk of data (structured dtype, ND flattening, sampling) + # Process a chunk of data (structured dtype, ND flattening) def process_data_chunk(data): data = structured_to_records(data) @@ -77,11 +64,11 @@ def process_data_chunk(data): return # Fast-path for simple 1D arrays - if isinstance(data, np.ndarray) and data.ndim == 1: + if isinstance(data, np.ndarray) and getattr(data, "ndim", 1) == 1: for val in data: try: row_val = convert_numpy_types(val) - add_row({"value": row_val}) + rows.append({"value": row_val}) except Exception: continue return @@ -99,7 +86,7 @@ def process_data_chunk(data): for _, row in df.iterrows(): try: row_dict = convert_numpy_types(row.to_dict()) - add_row(row_dict) + rows.append(row_dict) except Exception: try: basic_row = {} @@ -109,28 +96,28 @@ def process_data_chunk(data): basic_row[str(col)] = val.item() else: basic_row[str(col)] = str(val) - add_row(basic_row) + rows.append(basic_row) except Exception: continue else: try: val = convert_numpy_types(data) - add_row({"value": val}) + rows.append({"value": val}) except Exception: try: if hasattr(data, "item"): - add_row({"value": data.item()}) + rows.append({"value": data.item()}) else: - add_row({"value": str(data)}) + rows.append({"value": str(data)}) except Exception: return - # Chunked dataset reading + sampling + # Chunked dataset reading def recurse(name, obj, path=[]): if isinstance(obj, h5py.Dataset): shape = getattr(obj, "shape", None) - # Chunk mode for any large dataset (Option 2) + # Still chunk large datasets to avoid memory explosions if shape is not None and len(shape) > 0 and shape[0] > 10000: step = 10000 for i in range(0, shape[0], step): From b31e18df686644f53155f811a906a7f2ba90874d Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 30 Nov 2025 11:16:46 -0500 Subject: [PATCH 03/12] Refactor completeness function for chunk processing Refactor completeness function to process DataFrame in chunks, calculate completeness scores, and generate a visualization. Added error handling for file reading and completeness calculation. --- .../structured_data_metrics/completeness.py | 105 ++++++++++++------ 1 file changed, 70 insertions(+), 35 deletions(-) diff --git a/aidrin/structured_data_metrics/completeness.py b/aidrin/structured_data_metrics/completeness.py index 9e284534..a8156cd2 100644 --- a/aidrin/structured_data_metrics/completeness.py +++ b/aidrin/structured_data_metrics/completeness.py @@ -2,69 +2,104 @@ import io import matplotlib.pyplot as plt +import numpy as np from celery import Task, shared_task from celery.exceptions import SoftTimeLimitExceeded from aidrin.file_handling.file_parser import read_file +def iterate_chunks(df, chunksize=50000): + """Yield DataFrame chunks for any file type.""" + for start in range(0, len(df), chunksize): + yield df.iloc[start:start + chunksize] + + @shared_task(bind=True, ignore_result=False) def completeness(self: Task, file_info): try: - file = read_file(file_info) + # Reads the file + df = read_file(file_info) + + if df is None: + return {"Error": "File could not be read."} + + # Ensure column names are strings + if hasattr(df, 'columns'): + df.columns = [str(col) for col in df.columns] + + # Processes each chunk of rows + chunk_stats = [] + total_rows = 0 + + for idx, chunk in enumerate(iterate_chunks(df)): + chunk_size = len(chunk) + total_rows += chunk_size - # Ensure DataFrame columns are strings to avoid numpy array issues - if hasattr(file, 'columns'): - file.columns = [str(col) for col in file.columns] + # Missing values per column for this chunk + chunk_missing = {col: chunk[col].isnull().sum() for col in df.columns} - # Calculate completeness metric for each column - try: - completeness_scores = (1 - file.isnull().mean()).to_dict() - except Exception: - # If to_dict() fails, manually create the dictionary - completeness_scores = {} - for col in file.columns: - try: - completeness_scores[str(col)] = float(1 - file[col].isnull().mean()) - except Exception: - completeness_scores[str(col)] = 0.0 + # Missing in any column for this chunk + chunk_missing_rows = chunk.isnull().any(axis=1).sum() - # Ensure all column names are strings to avoid numpy array issues - completeness_scores = {str(k): v for k, v in completeness_scores.items()} + # Column-wise completeness for this chunk + chunk_completeness = { + col: 1 - (chunk_missing[col] / chunk_size) + for col in df.columns + } - # Calculate overall completeness metric for the dataset - overall_completeness = 1 - file.isnull().any(axis=1).mean() + # Overall completeness for this chunk + chunk_overall = 1 - (chunk_missing_rows / chunk_size) - result_dict = {} + chunk_stats.append({ + "chunk": idx, + "size": chunk_size, + "completeness": chunk_completeness, + "overall": chunk_overall + }) - # Always include completeness scores for all features - result_dict["Completeness scores"] = completeness_scores - result_dict["Overall Completeness"] = overall_completeness + # Aggregates completeness across all chunks + final_feature_scores = {} - # Create a bar chart for all features + for col in df.columns: + weighted_sum = sum( + cs["completeness"][col] * cs["size"] for cs in chunk_stats + ) + final_feature_scores[col] = weighted_sum / total_rows + + final_overall = sum( + cs["overall"] * cs["size"] for cs in chunk_stats + ) / total_rows + + # Creates the completeness chart plt.figure(figsize=(8, 6)) - plt.bar(completeness_scores.keys(), completeness_scores.values(), color="blue") - plt.title("Feature-wise Completeness Scores", fontsize=16) + plt.bar(final_feature_scores.keys(), final_feature_scores.values()) + plt.title("Final Feature-wise Completeness Scores", fontsize=16) plt.xlabel("Features", fontsize=14) plt.ylabel("Completeness Score", fontsize=14) plt.ylim(0, 1) - - # Rotate x-axis tick labels for readability plt.xticks(rotation=45, ha="right", fontsize=12) plt.tight_layout() - # Save the chart to a BytesIO object + # Converts chart to base64 img_buf = io.BytesIO() plt.savefig(img_buf, format="png") img_buf.seek(0) - - # Encode the image as base64 - img_base64 = base64.b64encode(img_buf.read()).decode("utf-8") - - result_dict["Completeness Visualization"] = img_base64 + final_img = base64.b64encode(img_buf.read()).decode("utf-8") plt.close() - return result_dict + # Returns all results + return { + "Chunk Completeness": chunk_stats, + "Final Completeness": { + "feature_wise": final_feature_scores, + "overall": final_overall, + }, + "Completeness Visualization": final_img + } except SoftTimeLimitExceeded: raise Exception("Completeness task timed out.") + + except Exception as e: + return {"Error": f"Completeness calculation failed: {str(e)}"} From 51cb023b830acf7d04c4d903e2af3bd7f29befa4 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 30 Nov 2025 11:18:06 -0500 Subject: [PATCH 04/12] Refactor outlier detection to use DataFrame chunks --- aidrin/structured_data_metrics/outliers.py | 163 +++++++++++++-------- 1 file changed, 98 insertions(+), 65 deletions(-) diff --git a/aidrin/structured_data_metrics/outliers.py b/aidrin/structured_data_metrics/outliers.py index 945ac31e..8a16daca 100644 --- a/aidrin/structured_data_metrics/outliers.py +++ b/aidrin/structured_data_metrics/outliers.py @@ -9,89 +9,122 @@ from aidrin.file_handling.file_parser import read_file +def iterate_chunks(df, chunksize=50000): + """Yield DataFrame chunks for any file type.""" + for start in range(0, len(df), chunksize): + yield df.iloc[start:start + chunksize] + + @shared_task(bind=True, ignore_result=False) def outliers(self: Task, file_info): try: - file = read_file(file_info) - - # Ensure DataFrame columns are strings to avoid numpy array issues - if hasattr(file, 'columns'): - file.columns = [str(col) for col in file.columns] - - try: - out_dict = {} - # Select numerical columns for outlier detection - numerical_columns = file.select_dtypes(include=[np.number]) - - if numerical_columns.empty: - return {"Error": "No numerical features found in the dataset."} + # Reads the file + df = read_file(file_info) - proportions_dict = {} + if df is None: + return {"Error": "File could not be read."} - # Process each column separately - for col in numerical_columns.columns: - series = numerical_columns[col].dropna() + # Ensures column names are strings + df.columns = [str(col) for col in df.columns] - if series.empty: - proportions_dict[col] = np.nan - continue - - q1 = series.quantile(0.25) - q3 = series.quantile(0.75) - IQR = q3 - q1 + # Selects only numeric columns + numeric_df = df.select_dtypes(include=[np.number]) - if IQR == 0: - proportions_dict[col] = 0.0 # no variability, no outliers - continue + if numeric_df.empty: + return {"Error": "No numerical features found in the dataset."} - # Identify outliers using IQR - mask = (series < (q1 - 1.5 * IQR)) | (series > (q3 + 1.5 * IQR)) - proportions_dict[col] = mask.mean() # proportion of outliers + numeric_cols = list(numeric_df.columns) - # Calculate overall outlier score - valid_values = [v for v in proportions_dict.values() if not np.isnan(v)] - overall_score = np.mean(valid_values) if valid_values else 0.0 - proportions_dict["Overall outlier score"] = overall_score + # Collects values for global quantile computation + collected = {col: [] for col in numeric_cols} - # Ensure all column names are strings to avoid numpy array issues - proportions_dict = {str(k): v for k, v in proportions_dict.items()} + for chunk in iterate_chunks(numeric_df): + for col in numeric_cols: + collected[col].extend(chunk[col].dropna().tolist()) - # Calculate the average of dictionary values - average_value = sum(proportions_dict.values()) / len(proportions_dict) - proportions_dict["Overall outlier score"] = average_value - # add the average to dictionary - out_dict["Outlier scores"] = proportions_dict + # Handles fully empty columns + for col in numeric_cols: + if len(collected[col]) == 0: + collected[col] = [np.nan] - # Create bar chart for feature-level outlier proportions only - feature_scores = { - k: v for k, v in proportions_dict.items() if k != "Overall outlier score" - } + # Computes Q1, Q3, and IQR for each column + stats = {} + for col in numeric_cols: + series = np.array(collected[col]) + if np.all(np.isnan(series)): + stats[col] = (np.nan, np.nan, np.nan) + continue - if feature_scores: # only plot if there are valid features - plt.figure(figsize=(8, 8)) - plt.bar(feature_scores.keys(), feature_scores.values(), color="red") - plt.title("Proportion of Outliers for Numerical Columns", fontsize=14) - plt.xlabel("Columns", fontsize=14) - plt.ylabel("Proportion of Outliers", fontsize=14) - plt.ylim(0, 1) + q1 = np.nanpercentile(series, 25) + q3 = np.nanpercentile(series, 75) + IQR = q3 - q1 + stats[col] = (q1, q3, IQR) - plt.xticks(rotation=45, ha="right", fontsize=12) - plt.subplots_adjust(bottom=0.5) - plt.tight_layout() + # Counts outliers across all chunks + total_counts = {col: 0 for col in numeric_cols} + outlier_counts = {col: 0 for col in numeric_cols} - # Save the chart to BytesIO and encode as base64 - img_buf = io.BytesIO() - plt.savefig(img_buf, format="png") - img_buf.seek(0) - img_base64 = base64.b64encode(img_buf.read()).decode("utf-8") + for chunk in iterate_chunks(numeric_df): + for col in numeric_cols: + col_data = chunk[col].dropna() + total_counts[col] += len(col_data) - out_dict["Outliers Visualization"] = img_base64 - plt.close() + q1, q3, IQR = stats[col] - return out_dict + if np.isnan(IQR) or IQR == 0: + continue - except Exception as e: - return {"Error": f"Outlier detection failed: {str(e)}"} + lower = q1 - 1.5 * IQR + upper = q3 + 1.5 * IQR + mask = (col_data < lower) | (col_data > upper) + outlier_counts[col] += mask.sum() + + # Computes final outlier proportions + proportions = {} + for col in numeric_cols: + if total_counts[col] == 0: + proportions[col] = 0.0 + else: + proportions[col] = outlier_counts[col] / total_counts[col] + + # Computes overall outlier score + valid_scores = [v for v in proportions.values() if not np.isnan(v)] + overall_score = float(np.mean(valid_scores)) if valid_scores else 0.0 + + proportions["Overall outlier score"] = overall_score + + # Builds response dictionary + out_dict = {"Outlier scores": proportions} + + # Creates visualization for feature-level outlier proportions + feature_scores = { + k: v for k, v in proportions.items() + if k != "Overall outlier score" + } + + if feature_scores: + plt.figure(figsize=(8, 8)) + plt.bar(feature_scores.keys(), feature_scores.values(), color="red") + plt.title("Proportion of Outliers for Numerical Columns", fontsize=14) + plt.xlabel("Columns", fontsize=14) + plt.ylabel("Proportion of Outliers", fontsize=14) + plt.ylim(0, 1) + plt.xticks(rotation=45, ha="right", fontsize=12) + plt.tight_layout() + + # Converts chart to base64 + img_buf = io.BytesIO() + plt.savefig(img_buf, format="png") + img_buf.seek(0) + img_base64 = base64.b64encode(img_buf.read()).decode("utf-8") + plt.close() + + out_dict["Outliers Visualization"] = img_base64 + + return out_dict except SoftTimeLimitExceeded: raise Exception("Outliers task timed out.") + except Exception as e: + return {"Error": f"Outlier detection failed: {str(e)}"} + From 366c6a3d176f6934daa2f582a7cdac2ff7f933e7 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 30 Nov 2025 11:18:27 -0500 Subject: [PATCH 05/12] Refactor duplicity task to handle DataFrame chunks --- aidrin/structured_data_metrics/duplicity.py | 71 +++++++++++++++++++-- 1 file changed, 64 insertions(+), 7 deletions(-) diff --git a/aidrin/structured_data_metrics/duplicity.py b/aidrin/structured_data_metrics/duplicity.py index c43701c6..516b6c28 100644 --- a/aidrin/structured_data_metrics/duplicity.py +++ b/aidrin/structured_data_metrics/duplicity.py @@ -1,21 +1,78 @@ from celery import Task, shared_task from celery.exceptions import SoftTimeLimitExceeded +import numpy as np from aidrin.file_handling.file_parser import read_file +def iterate_chunks(df, chunksize=50000): + """Yield DataFrame chunks for any file type.""" + for start in range(0, len(df), chunksize): + yield df.iloc[start:start + chunksize] + + +def row_to_key(row): + """ + Convert a pandas Series row into a hashable, NaN-normalized tuple. + + - NaNs are mapped to a sentinel so that rows with NaNs in the same positions + compare equal (matching pandas.duplicated behavior). + """ + sentinel = "__NAN__" + values = [] + for v in row: + # Treats all NaN-like values as the same + if isinstance(v, float) and np.isnan(v): + values.append(sentinel) + else: + values.append(v) + return tuple(values) + + @shared_task(bind=True, ignore_result=False) def duplicity(self: Task, file_info): try: - file = read_file(file_info) - dup_dict = {} - # Calculate the proportion of duplicate values - duplicate_proportions = file.duplicated().sum() / len(file) + df = read_file(file_info) + + if df is None: + return {"Error": "File could not be read."} + + # Ensure column names are strings + if hasattr(df, "columns"): + df.columns = [str(col) for col in df.columns] + + total_duplicates = 0 + total_rows = 0 + + # Global set to track unique row signatures across all chunks + global_seen = set() + + for chunk in iterate_chunks(df): + chunk_size = len(chunk) + if chunk_size == 0: + continue + + total_rows += chunk_size + + # Compute a normalized, hashable key for each row + # This mirrors pandas.duplicated's "NA values are equal" behavior. + for _, row in chunk.iterrows(): + key = row_to_key(row) + if key in global_seen: + total_duplicates += 1 + else: + global_seen.add(key) + + # Avoid division by zero + dup_score = total_duplicates / total_rows if total_rows > 0 else 0.0 - dup_dict["Duplicity scores"] = { - "Overall duplicity of the dataset": duplicate_proportions + return { + "Duplicity scores": { + "Overall duplicity of the dataset": dup_score + } } - return dup_dict except SoftTimeLimitExceeded: raise Exception("Duplicity task timed out.") + except Exception as e: + return {"Error": f"Duplicity detection failed: {str(e)}"} From 0efb7c144bbd1224eb472c03ea51f4ff03ded9b5 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:44:18 -0500 Subject: [PATCH 06/12] Update hdf5_reader.py --- aidrin/file_handling/readers/hdf5_reader.py | 242 ++++++++++---------- 1 file changed, 116 insertions(+), 126 deletions(-) diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py index a25cf832..0287ef49 100644 --- a/aidrin/file_handling/readers/hdf5_reader.py +++ b/aidrin/file_handling/readers/hdf5_reader.py @@ -1,6 +1,6 @@ import os import uuid -import random +import json import h5py import pandas as pd import numpy as np @@ -12,148 +12,136 @@ class hdf5Reader(BaseFileReader): def read(self): try: - rows = [] - - # Clean up byte strings in all object columns - def decode_bytes(df): - for col in df.columns: - if df[col].dtype == object: - df[col] = df[col].apply( - lambda x: x.decode("utf-8") - if isinstance(x, (bytes, bytearray, np.bytes_)) - else x - ) + CHUNK = 10_000 + MAX_TOTAL_ROWS = None + MAX_ROWS_PER_DATASET = None + + dfs = [] + total_rows = 0 + + def decode_bytes_inplace(df): + if df is None or df.empty: + return df + obj_cols = df.select_dtypes(include=["object"]).columns + for col in obj_cols: + df[col] = df[col].map( + lambda x: x.decode("utf-8") + if isinstance(x, (bytes, bytearray, np.bytes_)) + else x + ) return df - # Expand structured numpy dtypes into list of dicts - def structured_to_records(data): - try: - if isinstance(data, np.ndarray): - if getattr(data, "dtype", None) is not None and data.dtype.names: - return [dict(zip(data.dtype.names, row)) for row in data] - except Exception: - pass - return data - - # Convert numpy types to Python native types - def convert_numpy_types(obj): - try: - if hasattr(obj, "item"): - return obj.item() - elif isinstance(obj, (list, tuple)): - return [convert_numpy_types(x) for x in obj] - elif isinstance(obj, dict): - return {str(k): convert_numpy_types(v) for k, v in obj.items()} - elif isinstance(obj, np.ndarray): - if obj.size == 1: - return obj.item() - return [convert_numpy_types(x) for x in obj.tolist()] - return obj - except Exception: - return str(obj) - - # Process a chunk of data (structured dtype, ND flattening) - def process_data_chunk(data): - data = structured_to_records(data) - - # ND flattening for multidimensional arrays - if isinstance(data, np.ndarray) and hasattr(data, "ndim") and data.ndim > 2: - try: - data = np.ascontiguousarray(data).reshape(data.shape[0], -1) - except Exception: - return + def make_cells_hashable_inplace(df): + if df is None or df.empty: + return df - # Fast-path for simple 1D arrays - if isinstance(data, np.ndarray) and getattr(data, "ndim", 1) == 1: - for val in data: + def to_hashable(x): + if isinstance(x, (list, dict, set, tuple, np.ndarray)): try: - row_val = convert_numpy_types(val) - rows.append({"value": row_val}) + return json.dumps(x, default=str, ensure_ascii=False) except Exception: - continue - return + return str(x) + return x + + obj_cols = df.select_dtypes(include=["object"]).columns + for col in obj_cols: + df[col] = df[col].map(to_hashable) + return df + + def df_from_any(data): + if isinstance(data, np.ndarray): + if getattr(data.dtype, "names", None): + return pd.DataFrame.from_records(data) + + if data.ndim == 0: + return pd.DataFrame( + [{"value": data.item() if isinstance(data, np.generic) else data}] + ) + + if data.ndim == 1: + return pd.DataFrame({"value": data}) - # Handle table-like datasets - if isinstance(data, (list, tuple)) or hasattr(data, "dtype"): + if data.ndim > 2: + data = np.ascontiguousarray(data).reshape(data.shape[0], -1) + + if data.ndim == 2: + return pd.DataFrame(data) + + return pd.DataFrame({"value": [str(data)]}) + + if isinstance(data, (list, tuple)): try: - df = pd.DataFrame(data) + return pd.DataFrame(data) except Exception: - try: - df = pd.DataFrame(data.tolist()) - except Exception: - return + return pd.DataFrame({"value": list(data)}) - for _, row in df.iterrows(): - try: - row_dict = convert_numpy_types(row.to_dict()) - rows.append(row_dict) - except Exception: - try: - basic_row = {} - for col in row.index: - val = row[col] - if hasattr(val, "item"): - basic_row[str(col)] = val.item() - else: - basic_row[str(col)] = str(val) - rows.append(basic_row) - except Exception: - continue - else: + return pd.DataFrame([{"value": data}]) + + def add_df(df): + nonlocal total_rows + if df is None or df.empty: + return + + df.columns = [str(c) for c in df.columns] + + df = decode_bytes_inplace(df) + df = make_cells_hashable_inplace(df) + + dfs.append(df) + total_rows += len(df) + + with h5py.File(self.file_path, "r") as f: + + def handle_dataset(name, dset): + nonlocal total_rows try: - val = convert_numpy_types(data) - rows.append({"value": val}) - except Exception: - try: - if hasattr(data, "item"): - rows.append({"value": data.item()}) - else: - rows.append({"value": str(data)}) - except Exception: + shape = getattr(dset, "shape", None) + dtype = getattr(dset, "dtype", None) + self.logger.info(f"HDF5 dataset: {name} shape={shape} dtype={dtype}") + + if shape is None or len(shape) == 0: + df = df_from_any(dset[()]) + add_df(df) return - # Chunked dataset reading - def recurse(name, obj, path=[]): - if isinstance(obj, h5py.Dataset): - shape = getattr(obj, "shape", None) - - # Still chunk large datasets to avoid memory explosions - if shape is not None and len(shape) > 0 and shape[0] > 10000: - step = 10000 - for i in range(0, shape[0], step): - try: - chunk = obj[i:i+step] - process_data_chunk(chunk) - except Exception: - break - else: - try: - data = obj[()] - process_data_chunk(data) - except Exception: + n = int(shape[0]) if shape[0] is not None else 0 + if n == 0: + df = df_from_any(dset[()]) + add_df(df) return - # Traverse all datasets - with h5py.File(self.file_path, "r") as f: - def visit(name, obj): - recurse(name, obj, name.strip("/").split("/")) - f.visititems(visit) + limit = n + if isinstance(MAX_ROWS_PER_DATASET, int): + limit = min(limit, MAX_ROWS_PER_DATASET) + + for start in range(0, limit, CHUNK): + if isinstance(MAX_TOTAL_ROWS, int) and total_rows >= MAX_TOTAL_ROWS: + self.logger.warning("Stopping early due to MAX_TOTAL_ROWS cap") + return + + chunk = dset[start: start + CHUNK] + df = df_from_any(chunk) + add_df(df) - # Create final DataFrame - df = pd.DataFrame(rows) - df = decode_bytes(df) + except Exception: + self.logger.exception(f"Failed reading dataset: {name}") + + def visitor(name, obj): + if isinstance(obj, h5py.Dataset): + handle_dataset(name, obj) - if hasattr(df, "columns") and len(df.columns) > 0: - df.columns = [str(col) for col in df.columns] + f.visititems(visitor) - if df.empty: + if not dfs: self.logger.warning("No data was successfully processed from HDF5 file") return None - return df + out = pd.concat(dfs, ignore_index=True, sort=False) + return None if out.empty else out except Exception as e: self.logger.error(f"Error while reading: {e}") + self.logger.exception("HDF5 read() failed with exception") return None def parse(self): @@ -166,10 +154,11 @@ def recurse(data): if isinstance(obj, h5py.Group): group_names.append(full_path) recurse(obj) - except (TypeError, ValueError): + except (TypeError, ValueError) as e: + self.logger.warning(f"Skipping unhashable key {name}: {e}") continue - except Exception: - pass + except Exception as e: + self.logger.error(f"Error during recursion: {e}") return group_names with h5py.File(self.file_path, "r") as f: @@ -191,9 +180,7 @@ def filter(self, kept_keys): self.logger.warning(f"Skipping unhashable key {g}: {e}") continue - new_file_name = ( - f"filtered_{uuid.uuid4().hex}_{session.get('uploaded_file_name')}" - ) + new_file_name = f"filtered_{uuid.uuid4().hex}_{session.get('uploaded_file_name')}" new_file_path = os.path.join(current_app.config["UPLOAD_FOLDER"], new_file_name) with ( @@ -204,12 +191,14 @@ def filter(self, kept_keys): def copy_group(path, src_group, tgt_group): for name, obj in src_group.items(): full_path = f"{path}/{name}".strip("/") + if isinstance(obj, h5py.Group): if full_path in filtered_keys: tgt_subgroup = tgt_group.create_group(name) copy_group(full_path, obj, tgt_subgroup) else: copy_group(full_path, obj, tgt_group) + elif isinstance(obj, h5py.Dataset): if path.strip("/") in filtered_keys: tgt_group.create_dataset(name, data=obj[()]) @@ -218,3 +207,4 @@ def copy_group(path, src_group, tgt_group): return new_file_path + From 21abf254ba18ca3d456c83dd73377b91940190c9 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:44:43 -0500 Subject: [PATCH 07/12] Remove MAX_TOTAL_ROWS and MAX_ROWS_PER_DATASET Removed unused constants for maximum rows. --- aidrin/file_handling/readers/hdf5_reader.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py index 0287ef49..11b62564 100644 --- a/aidrin/file_handling/readers/hdf5_reader.py +++ b/aidrin/file_handling/readers/hdf5_reader.py @@ -13,8 +13,7 @@ class hdf5Reader(BaseFileReader): def read(self): try: CHUNK = 10_000 - MAX_TOTAL_ROWS = None - MAX_ROWS_PER_DATASET = None + dfs = [] total_rows = 0 From 8f5f55ef8fb554d5b4dbe8be511ee1cf8c3a591e Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:45:27 -0500 Subject: [PATCH 08/12] Update outliers.py --- aidrin/structured_data_metrics/outliers.py | 270 ++++++++++++++------- 1 file changed, 178 insertions(+), 92 deletions(-) diff --git a/aidrin/structured_data_metrics/outliers.py b/aidrin/structured_data_metrics/outliers.py index 8a16daca..e519fbb4 100644 --- a/aidrin/structured_data_metrics/outliers.py +++ b/aidrin/structured_data_metrics/outliers.py @@ -1,130 +1,216 @@ import base64 import io +from typing import Optional +import matplotlib + +matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np +import pandas as pd from celery import Task, shared_task from celery.exceptions import SoftTimeLimitExceeded from aidrin.file_handling.file_parser import read_file -def iterate_chunks(df, chunksize=50000): - """Yield DataFrame chunks for any file type.""" - for start in range(0, len(df), chunksize): - yield df.iloc[start:start + chunksize] +MAX_B64_LEN = 250_000 +MAX_BARS = 40 + + +def _fig_to_b64(fig: plt.Figure, dpi: int = 85) -> str: + """Convert a specific Matplotlib Figure to base64 PNG.""" + buf = io.BytesIO() + fig.savefig(buf, format="png", dpi=dpi, bbox_inches="tight") + buf.seek(0) + s = base64.b64encode(buf.read()).decode("utf-8") + plt.close(fig) + return s + + +def _info_image(message: str) -> str: + fig = plt.figure(figsize=(7.5, 2.3)) + ax = fig.add_subplot(111) + ax.axis("off") + ax.text(0.01, 0.5, message, fontsize=11, va="center") + fig.tight_layout() + return _fig_to_b64(fig, dpi=95) + + +def _barplot_small( + keys, + vals, + title: str, + max_bars: int = MAX_BARS, + top: bool = True, +) -> Optional[str]: + keys = list(keys) + vals = np.asarray(list(vals), dtype=float) + + mask = np.isfinite(vals) + keys = [k for k, m in zip(keys, mask) if m] + vals = vals[mask] + if vals.size == 0: + return None + + if len(keys) > max_bars: + order = np.argsort(vals) + order = order[-max_bars:] if top else order[:max_bars] + keys = [keys[i] for i in order] + vals = vals[order] + + x = np.arange(len(keys)) + fig = plt.figure(figsize=(7.5, 3.2)) + ax = fig.add_subplot(111) + ax.bar(x, vals) + ax.set_title(title, fontsize=12) + ax.set_ylim(0, 1) + + step = max(1, len(keys) // 10) + tick_idx = np.arange(0, len(keys), step) + ax.set_xticks(tick_idx) + ax.set_xticklabels( + [str(keys[i]) for i in tick_idx], + rotation=45, + ha="right", + fontsize=8, + ) + + fig.tight_layout() + return _fig_to_b64(fig, dpi=85) @shared_task(bind=True, ignore_result=False) def outliers(self: Task, file_info): + """ + Exact IQR outlier proportions per numeric column (matches OG semantics): + - For each numeric col: compute q1/q3 on non-null values + - Outlier mask on those same non-null values using 1.5*IQR fences + - Proportion = (# outliers) / (# non-null) + - Overall outlier score = mean of per-feature proportions (finite only) + Backward-compatible output keys: + - "Outlier scores" dict includes "Overall outlier score" + - "Outliers Visualization" is base64 PNG when possible + """ try: - # Reads the file df = read_file(file_info) if df is None: - return {"Error": "File could not be read."} - - # Ensures column names are strings - df.columns = [str(col) for col in df.columns] - - # Selects only numeric columns - numeric_df = df.select_dtypes(include=[np.number]) - - if numeric_df.empty: - return {"Error": "No numerical features found in the dataset."} - - numeric_cols = list(numeric_df.columns) + return { + "Error": "File could not be read.", + "Outliers Visualization": _info_image( + "Outliers unavailable: file could not be read." + ), + } + + if not hasattr(df, "columns") or df.empty: + return { + "Error": "Dataset is empty.", + "Outliers Visualization": _info_image( + "Outliers unavailable: dataset is empty." + ), + } + + df.columns = [str(c) for c in df.columns] + + + for c in df.columns: + if df[c].dtype == object: + coerced = pd.to_numeric(df[c], errors="coerce") + + if np.isfinite(coerced.to_numpy(dtype=float, copy=False)).sum() > 0: + df[c] = coerced + + numerical_df = df.select_dtypes(include=[np.number]) + if numerical_df.empty: + + return { + "Error": "No numerical features found in the dataset.", + "Outlier scores": {"Overall outlier score": 0.0}, + "Outliers Visualization": _info_image( + "No numerical features found in the dataset." + ), + } - # Collects values for global quantile computation - collected = {col: [] for col in numeric_cols} - - for chunk in iterate_chunks(numeric_df): - for col in numeric_cols: - collected[col].extend(chunk[col].dropna().tolist()) + proportions = {} - # Handles fully empty columns - for col in numeric_cols: - if len(collected[col]) == 0: - collected[col] = [np.nan] + for col in numerical_df.columns: + series = numerical_df[col].dropna() - # Computes Q1, Q3, and IQR for each column - stats = {} - for col in numeric_cols: - series = np.array(collected[col]) - if np.all(np.isnan(series)): - stats[col] = (np.nan, np.nan, np.nan) + if series.empty: + proportions[str(col)] = np.nan continue - q1 = np.nanpercentile(series, 25) - q3 = np.nanpercentile(series, 75) - IQR = q3 - q1 - stats[col] = (q1, q3, IQR) - - # Counts outliers across all chunks - total_counts = {col: 0 for col in numeric_cols} - outlier_counts = {col: 0 for col in numeric_cols} - - for chunk in iterate_chunks(numeric_df): - for col in numeric_cols: - col_data = chunk[col].dropna() - total_counts[col] += len(col_data) + q1 = float(series.quantile(0.25)) + q3 = float(series.quantile(0.75)) + iqr = float(q3 - q1) - q1, q3, IQR = stats[col] - - if np.isnan(IQR) or IQR == 0: - continue - - lower = q1 - 1.5 * IQR - upper = q3 + 1.5 * IQR - mask = (col_data < lower) | (col_data > upper) - outlier_counts[col] += mask.sum() - - # Computes final outlier proportions - proportions = {} - for col in numeric_cols: - if total_counts[col] == 0: - proportions[col] = 0.0 - else: - proportions[col] = outlier_counts[col] / total_counts[col] + if not np.isfinite(iqr) or iqr == 0.0: + proportions[str(col)] = 0.0 + continue - # Computes overall outlier score - valid_scores = [v for v in proportions.values() if not np.isnan(v)] - overall_score = float(np.mean(valid_scores)) if valid_scores else 0.0 + lower = q1 - 1.5 * iqr + upper = q3 + 1.5 * iqr + mask = (series < lower) | (series > upper) + proportions[str(col)] = float(mask.mean()) + valid_values = [v for v in proportions.values() if np.isfinite(v)] + overall_score = float(np.mean(valid_values)) if valid_values else 0.0 proportions["Overall outlier score"] = overall_score - # Builds response dictionary - out_dict = {"Outlier scores": proportions} + out = {"Outlier scores": proportions} - # Creates visualization for feature-level outlier proportions + # Visualization (feature-level only) feature_scores = { - k: v for k, v in proportions.items() - if k != "Overall outlier score" + k: v for k, v in proportions.items() if k != "Overall outlier score" } - if feature_scores: - plt.figure(figsize=(8, 8)) - plt.bar(feature_scores.keys(), feature_scores.values(), color="red") - plt.title("Proportion of Outliers for Numerical Columns", fontsize=14) - plt.xlabel("Columns", fontsize=14) - plt.ylabel("Proportion of Outliers", fontsize=14) - plt.ylim(0, 1) - plt.xticks(rotation=45, ha="right", fontsize=12) - plt.tight_layout() - - # Converts chart to base64 - img_buf = io.BytesIO() - plt.savefig(img_buf, format="png") - img_buf.seek(0) - img_base64 = base64.b64encode(img_buf.read()).decode("utf-8") - plt.close() - - out_dict["Outliers Visualization"] = img_base64 - - return out_dict + if not feature_scores: + out["Outliers Visualization"] = _info_image( + "Outliers computed, but no feature scores to plot." + ) + return out + + vals = np.asarray(list(feature_scores.values()), dtype=float) + finite_vals = vals[np.isfinite(vals)] + + if finite_vals.size == 0: + out["Outliers Visualization"] = _info_image( + "Outliers computed, but no finite scores to plot." + ) + return out + + if float(np.nanmax(finite_vals)) == 0.0: + out["Outliers Visualization"] = _info_image( + "No outliers detected (all proportions are 0).\n" + "Common causes: very few rows, constant columns (IQR=0)." + ) + return out + + img = _barplot_small( + keys=list(feature_scores.keys()), + vals=list(feature_scores.values()), + title="Proportion of Outliers (top features)", + max_bars=MAX_BARS, + top=True, + ) + + if (not img) or (len(img) > MAX_B64_LEN): + out["Outliers Visualization"] = _info_image( + "Outliers chart omitted to prevent oversized payload.\n" + "Scores are computed successfully." + ) + else: + out["Outliers Visualization"] = img + + return out except SoftTimeLimitExceeded: raise Exception("Outliers task timed out.") except Exception as e: - return {"Error": f"Outlier detection failed: {str(e)}"} + msg = f"Outlier detection failed: {str(e)}" + return { + "Error": msg, + "Outliers Visualization": _info_image(msg), + } From ea8cefa2bf3f9afb0a8b246a52495cbe7196df44 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:46:21 -0500 Subject: [PATCH 09/12] Remove docstring from outliers function Removed docstring explaining the outliers function. --- aidrin/structured_data_metrics/outliers.py | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/aidrin/structured_data_metrics/outliers.py b/aidrin/structured_data_metrics/outliers.py index e519fbb4..aa70e3e3 100644 --- a/aidrin/structured_data_metrics/outliers.py +++ b/aidrin/structured_data_metrics/outliers.py @@ -82,16 +82,7 @@ def _barplot_small( @shared_task(bind=True, ignore_result=False) def outliers(self: Task, file_info): - """ - Exact IQR outlier proportions per numeric column (matches OG semantics): - - For each numeric col: compute q1/q3 on non-null values - - Outlier mask on those same non-null values using 1.5*IQR fences - - Proportion = (# outliers) / (# non-null) - - Overall outlier score = mean of per-feature proportions (finite only) - Backward-compatible output keys: - - "Outlier scores" dict includes "Overall outlier score" - - "Outliers Visualization" is base64 PNG when possible - """ + try: df = read_file(file_info) From b512984c69f6e402de87ab2ac072a81e8527b253 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:46:44 -0500 Subject: [PATCH 10/12] Refactor completeness.py for improved readability --- .../structured_data_metrics/completeness.py | 237 +++++++++++++----- 1 file changed, 171 insertions(+), 66 deletions(-) diff --git a/aidrin/structured_data_metrics/completeness.py b/aidrin/structured_data_metrics/completeness.py index a8156cd2..9c026521 100644 --- a/aidrin/structured_data_metrics/completeness.py +++ b/aidrin/structured_data_metrics/completeness.py @@ -1,6 +1,10 @@ import base64 import io +from typing import Optional +import matplotlib + +matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np from celery import Task, shared_task @@ -9,97 +13,198 @@ from aidrin.file_handling.file_parser import read_file -def iterate_chunks(df, chunksize=50000): - """Yield DataFrame chunks for any file type.""" +def iterate_chunks(df, chunksize: int = 50_000): for start in range(0, len(df), chunksize): - yield df.iloc[start:start + chunksize] + yield df.iloc[start : start + chunksize] + + +MAX_B64_LEN = 250_000 +MAX_BARS = 40 +CHUNK_SIZE = 50_000 +MAX_CHUNK_STATS = 200 # cap response size + + +def _fig_to_b64(dpi: int = 85) -> str: + buf = io.BytesIO() + plt.savefig(buf, format="png", dpi=dpi, bbox_inches="tight") + buf.seek(0) + s = base64.b64encode(buf.read()).decode("utf-8") + plt.close() + return s + + +def _info_image(message: str) -> str: + plt.figure(figsize=(7.5, 2.3)) + plt.axis("off") + plt.text(0.01, 0.5, message, fontsize=11, va="center") + plt.tight_layout() + return _fig_to_b64(dpi=95) + + +def _barplot_small( + keys, + vals, + title: str, + max_bars: int = MAX_BARS, + worst: bool = True, +) -> Optional[str]: + keys = list(keys) + vals = np.asarray(list(vals), dtype=float) + + mask = np.isfinite(vals) + keys = [k for k, m in zip(keys, mask) if m] + vals = vals[mask] + if vals.size == 0: + return None + + if len(keys) > max_bars: + order = np.argsort(vals) + order = order[:max_bars] if worst else order[-max_bars:] + keys = [keys[i] for i in order] + vals = vals[order] + + x = np.arange(len(keys)) + plt.figure(figsize=(7.5, 3.2)) + plt.bar(x, vals) + plt.title(title, fontsize=12) + plt.ylim(0, 1) + + step = max(1, len(keys) // 10) + tick_idx = np.arange(0, len(keys), step) + plt.xticks( + tick_idx, + [str(keys[i]) for i in tick_idx], + rotation=45, + ha="right", + fontsize=8, + ) + + plt.tight_layout() + return _fig_to_b64(dpi=85) @shared_task(bind=True, ignore_result=False) def completeness(self: Task, file_info): try: - # Reads the file df = read_file(file_info) if df is None: - return {"Error": "File could not be read."} + return { + "Error": "File could not be read.", + "Completeness Visualization": _info_image( + "Completeness unavailable: file could not be read." + ), + } - # Ensure column names are strings - if hasattr(df, 'columns'): - df.columns = [str(col) for col in df.columns] + if not hasattr(df, "columns") or df.empty: + return { + "Error": "Dataset is empty.", + "Completeness Visualization": _info_image( + "Completeness unavailable: dataset is empty." + ), + } + + # Ensure column names are plain strings + df.columns = [str(c) for c in df.columns] + cols = list(df.columns) - # Processes each chunk of rows - chunk_stats = [] total_rows = 0 + total_missing_per_col = {col: 0 for col in cols} + total_rows_with_any_missing = 0 - for idx, chunk in enumerate(iterate_chunks(df)): - chunk_size = len(chunk) - total_rows += chunk_size + chunk_stats = [] - # Missing values per column for this chunk - chunk_missing = {col: chunk[col].isnull().sum() for col in df.columns} + for idx, chunk in enumerate(iterate_chunks(df, chunksize=CHUNK_SIZE)): + chunk_size = int(len(chunk)) + if chunk_size == 0: + continue - # Missing in any column for this chunk - chunk_missing_rows = chunk.isnull().any(axis=1).sum() + total_rows += chunk_size - # Column-wise completeness for this chunk - chunk_completeness = { - col: 1 - (chunk_missing[col] / chunk_size) - for col in df.columns + miss = chunk.isna().sum().to_dict() + for col in cols: + total_missing_per_col[col] += int(miss.get(col, 0)) + + rows_any_missing = int(chunk.isna().any(axis=1).sum()) + total_rows_with_any_missing += rows_any_missing + + + if idx < MAX_CHUNK_STATS: + chunk_overall = float(1.0 - (rows_any_missing / chunk_size)) + chunk_stats.append( + { + "chunk": idx, + "size": chunk_size, + "overall": chunk_overall, + } + ) + + if total_rows == 0: + return { + "Error": "No rows found in dataset.", + "Completeness Visualization": _info_image( + "Completeness unavailable: no rows found." + ), } - # Overall completeness for this chunk - chunk_overall = 1 - (chunk_missing_rows / chunk_size) - - chunk_stats.append({ - "chunk": idx, - "size": chunk_size, - "completeness": chunk_completeness, - "overall": chunk_overall - }) - - # Aggregates completeness across all chunks - final_feature_scores = {} - - for col in df.columns: - weighted_sum = sum( - cs["completeness"][col] * cs["size"] for cs in chunk_stats - ) - final_feature_scores[col] = weighted_sum / total_rows - - final_overall = sum( - cs["overall"] * cs["size"] for cs in chunk_stats - ) / total_rows - - # Creates the completeness chart - plt.figure(figsize=(8, 6)) - plt.bar(final_feature_scores.keys(), final_feature_scores.values()) - plt.title("Final Feature-wise Completeness Scores", fontsize=16) - plt.xlabel("Features", fontsize=14) - plt.ylabel("Completeness Score", fontsize=14) - plt.ylim(0, 1) - plt.xticks(rotation=45, ha="right", fontsize=12) - plt.tight_layout() - - # Converts chart to base64 - img_buf = io.BytesIO() - plt.savefig(img_buf, format="png") - img_buf.seek(0) - final_img = base64.b64encode(img_buf.read()).decode("utf-8") - plt.close() - - # Returns all results - return { + # Final feature-wise completeness + final_feature_scores = { + col: float(1.0 - (total_missing_per_col[col] / total_rows)) for col in cols + } + # Final overall completeness (row has no missing values) + final_overall = float(1.0 - (total_rows_with_any_missing / total_rows)) + + out = { + + "Completeness scores": final_feature_scores, + "Overall Completeness": final_overall, + "Chunk Completeness": chunk_stats, "Final Completeness": { "feature_wise": final_feature_scores, "overall": final_overall, }, - "Completeness Visualization": final_img } + vals = np.asarray(list(final_feature_scores.values()), dtype=float) + finite_vals = vals[np.isfinite(vals)] + + if finite_vals.size == 0: + out["Completeness Visualization"] = _info_image( + "Completeness computed, but no finite scores to plot." + ) + return out + + if float(np.nanmin(finite_vals)) == 1.0 and float(np.nanmax(finite_vals)) == 1.0: + out["Completeness Visualization"] = _info_image( + "All features are 100% complete.\nNo missing values detected." + ) + return out + + img = _barplot_small( + keys=list(final_feature_scores.keys()), + vals=list(final_feature_scores.values()), + title="Feature-wise Completeness (worst features)", + max_bars=MAX_BARS, + worst=True, + ) + + if (not img) or (len(img) > MAX_B64_LEN): + out["Completeness Visualization"] = _info_image( + "Completeness chart omitted to prevent oversized payload.\n" + "Scores are computed successfully." + ) + else: + out["Completeness Visualization"] = img + + return out + except SoftTimeLimitExceeded: raise Exception("Completeness task timed out.") - except Exception as e: - return {"Error": f"Completeness calculation failed: {str(e)}"} + msg = f"Completeness calculation failed: {str(e)}" + return { + "Error": msg, + "Completeness Visualization": _info_image(f"{msg}"), + } + From 6bc418f1e79b355469890f0e758960e4e9811ed3 Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:47:01 -0500 Subject: [PATCH 11/12] Refactor duplicity task for improved file handling --- aidrin/structured_data_metrics/duplicity.py | 98 +++++++++------------ 1 file changed, 44 insertions(+), 54 deletions(-) diff --git a/aidrin/structured_data_metrics/duplicity.py b/aidrin/structured_data_metrics/duplicity.py index 516b6c28..72d16b5f 100644 --- a/aidrin/structured_data_metrics/duplicity.py +++ b/aidrin/structured_data_metrics/duplicity.py @@ -1,78 +1,68 @@ from celery import Task, shared_task from celery.exceptions import SoftTimeLimitExceeded import numpy as np +import pandas as pd from aidrin.file_handling.file_parser import read_file -def iterate_chunks(df, chunksize=50000): - """Yield DataFrame chunks for any file type.""" - for start in range(0, len(df), chunksize): - yield df.iloc[start:start + chunksize] - - -def row_to_key(row): - """ - Convert a pandas Series row into a hashable, NaN-normalized tuple. - - - NaNs are mapped to a sentinel so that rows with NaNs in the same positions - compare equal (matching pandas.duplicated behavior). - """ - sentinel = "__NAN__" - values = [] - for v in row: - # Treats all NaN-like values as the same - if isinstance(v, float) and np.isnan(v): - values.append(sentinel) - else: - values.append(v) - return tuple(values) - - @shared_task(bind=True, ignore_result=False) def duplicity(self: Task, file_info): try: - df = read_file(file_info) - - if df is None: - return {"Error": "File could not be read."} + file = read_file(file_info) - # Ensure column names are strings - if hasattr(df, "columns"): - df.columns = [str(col) for col in df.columns] - - total_duplicates = 0 - total_rows = 0 + if file is None or not hasattr(file, "empty") or file.empty: + return { + "Duplicity scores": { + "Overall duplicity of the dataset": 0.0 + } + } - # Global set to track unique row signatures across all chunks - global_seen = set() + n_rows = len(file) + n_cols = len(file.columns) - for chunk in iterate_chunks(df): - chunk_size = len(chunk) - if chunk_size == 0: - continue + if n_rows >= 2: + dup = float(file.duplicated().sum() / n_rows) + return { + "Duplicity scores": { + "Overall duplicity of the dataset": dup + } + } - total_rows += chunk_size + numeric = file.select_dtypes(include=[np.number]) + + if numeric.empty: + vals = file.to_numpy().ravel() + vals = vals[~pd.isna(vals)] + total = int(vals.size) + if total <= 1: + dup = 0.0 + else: + uniq = int(pd.unique(vals).size) + dup = float(1.0 - (uniq / total)) + return { + "Duplicity scores": { + "Overall duplicity of the dataset": dup + } + } - # Compute a normalized, hashable key for each row - # This mirrors pandas.duplicated's "NA values are equal" behavior. - for _, row in chunk.iterrows(): - key = row_to_key(row) - if key in global_seen: - total_duplicates += 1 - else: - global_seen.add(key) + vals = numeric.to_numpy().ravel() + vals = vals[np.isfinite(vals)] + total = int(vals.size) - # Avoid division by zero - dup_score = total_duplicates / total_rows if total_rows > 0 else 0.0 + if total <= 1: + dup = 0.0 + else: + vals = np.round(vals, 6) + uniq = int(pd.unique(vals).size) + dup = float(1.0 - (uniq / total)) return { "Duplicity scores": { - "Overall duplicity of the dataset": dup_score + "Overall duplicity of the dataset": dup } } except SoftTimeLimitExceeded: raise Exception("Duplicity task timed out.") - except Exception as e: - return {"Error": f"Duplicity detection failed: {str(e)}"} + From b8987e6b034c5b9477ea979cedd69b1f3b062efe Mon Sep 17 00:00:00 2001 From: varunviswapriyan <150558463+varunviswapriyan@users.noreply.github.com> Date: Sun, 7 Dec 2025 19:51:05 -0500 Subject: [PATCH 12/12] Add MAX_TOTAL_ROWS and MAX_ROWS_PER_DATASET variables --- aidrin/file_handling/readers/hdf5_reader.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/aidrin/file_handling/readers/hdf5_reader.py b/aidrin/file_handling/readers/hdf5_reader.py index 11b62564..3d8fdf9b 100644 --- a/aidrin/file_handling/readers/hdf5_reader.py +++ b/aidrin/file_handling/readers/hdf5_reader.py @@ -14,6 +14,8 @@ def read(self): try: CHUNK = 10_000 + MAX_TOTAL_ROWS = None + MAX_ROWS_PER_DATASET = None dfs = [] total_rows = 0