ayushsahu45/Multi-AI-Analytics-Platform
1
1# import pandas as pd2# import pyarrow as pa3# import pyarrow.parquet as pq4# from pathlib import Path5# from typing import Dict, Any, List, Union6# import json7# from datetime import datetime8 9 10# class PowerBIExporter:11# def __init__(self, output_dir: Union[str, Path]):12# self.output_dir = Path(output_dir)13# self.output_dir.mkdir(parents=True, exist_ok=True)14# self.exported_files = []15 16# def export_to_csv(self, df: pd.DataFrame, filename: str) -> Path:17# output_path = self.output_dir / f"{filename}.csv"18# df.to_csv(output_path, index=False)19# self.exported_files.append(output_path)20# return output_path21 22# def export_to_parquet(self, df: pd.DataFrame, filename: str) -> Path:23# output_path = self.output_dir / f"{filename}.parquet"24# df.to_parquet(output_path, index=False, engine='pyarrow')25# self.exported_files.append(output_path)26# return output_path27 28# def export_to_json(self, data: Any, filename: str) -> Path:29# output_path = self.output_dir / f"{filename}.json"30 31# with open(output_path, 'w', encoding='utf-8') as f:32# json.dump(data, f, indent=2, default=str)33 34# self.exported_files.append(output_path)35# return output_path36 37# def create_data_model(self, tables: Dict[str, pd.DataFrame], relationships: List[Dict[str, str]] = None) -> Dict[str, Any]:38# data_model = {39# "tables": {},40# "relationships": relationships or [],41# "created_at": datetime.now().isoformat()42# }43 44# for table_name, df in tables.items():45# data_model["tables"][table_name] = {46# "columns": df.columns.tolist(),47# "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},48# "row_count": len(df),49# "primary_key": df.columns[0] if len(df.columns) > 0 else None50# }51 52# model_path = self.export_to_json(data_model, "powerbi_data_model")53# return data_model54 55# def create_analysis_results(self, ml_results: Dict[str, Any], dl_results: Dict[str, Any], 56# data_summary: Dict[str, Any]) -> pd.DataFrame:57# results_df = pd.DataFrame([58# {59# "metric_category": "Machine Learning",60# "metric_name": "accuracy" if "accuracy" in ml_results else "mse",61# "metric_value": ml_results.get("accuracy", ml_results.get("mse", 0)),62# "timestamp": datetime.now()63# },64# {65# "metric_category": "Deep Learning",66# "metric_name": "device",67# "metric_value": dl_results.get("device", "unknown"),68# "timestamp": datetime.now()69# },70# {71# "metric_category": "Data Summary",72# "metric_name": "row_count",73# "metric_value": data_summary.get("row_count", 0),74# "timestamp": datetime.now()75# }76# ])77 78# return results_df79 80# def export_predictions(self, df: pd.DataFrame, predictions: List[Any], 81# probabilities: List[List[float]] = None, filename: str = "predictions") -> Path:82# result_df = df.copy()83# result_df["prediction"] = predictions84 85# if probabilities:86# for i, probs in enumerate(zip(*probabilities)):87# result_df[f"prob_class_{i}"] = probs88 89# return self.export_to_csv(result_df, filename)90 91# def create_dashboard_data(self, analysis_results: Dict[str, Any]) -> Dict[str, pd.DataFrame]:92# dashboard_data = {}93 94# if "feature_importance" in analysis_results:95# dashboard_data["feature_importance"] = pd.DataFrame(analysis_results["feature_importance"])96 97# if "predictions" in analysis_results:98# dashboard_data["predictions"] = pd.DataFrame(analysis_results["predictions"])99 100# if "metrics" in analysis_results:101# metrics_list = []102# for key, value in analysis_results["metrics"].items():103# if isinstance(value, (int, float)):104# metrics_list.append({"metric": key, "value": value})105# if metrics_list:106# dashboard_data["metrics_summary"] = pd.DataFrame(metrics_list)107 108# return dashboard_data109 110# def export_all(self, dataframes: Dict[str, pd.DataFrame], include_parquet: bool = True) -> List[Path]:111# exported = []112 113# for name, df in dataframes.items():114# csv_path = self.export_to_csv(df, name)115# exported.append(csv_path)116 117# if include_parquet:118# parquet_path = self.export_to_parquet(df, name)119# exported.append(parquet_path)120 121# return exported122 123# def get_exported_files(self) -> List[Path]:124# return self.exported_files125 126# def generate_powerbi_instructions(self) -> str:127# instructions = """128# Power BI Integration Instructions:129# ================================130 131# 1. Open Power BI Desktop132 133# 2. Get Data:134# - Click "Get Data" > "More..."135# - Select "Text/CSV" for CSV files136# - Select "Parquet" for Parquet files137 138# 3. Load the exported data:139# - Navigate to the 'output' folder140# - Select the relevant CSV/Parquet files141 142# 4. Create relationships:143# - Open "Model" view144# - Drag columns to create relationships between tables145 146# 5. Build visualizations:147# - Use the "Visualizations" pane148# - Create charts, tables, and KPIs149 150# Exported files are located in: {output_dir}151# """.format(output_dir=str(self.output_dir))152 153# return instructions154 155 156 157 158 159 160import pandas as pd161from pathlib import Path162from typing import Dict, Any, List, Union, Optional163import json164from datetime import datetime165 166 167class PowerBIExporter:168 def __init__(self, output_dir: Union[str, Path]):169 self.output_dir = Path(output_dir)170 self.output_dir.mkdir(parents=True, exist_ok=True)171 self.exported_files: List[Path] = []172 173 def export_to_csv(self, df: pd.DataFrame, filename: str) -> Path:174 output_path = self.output_dir / f"{filename}.csv"175 df.to_csv(output_path, index=False)176 self.exported_files.append(output_path)177 return output_path178 179 def export_to_parquet(self, df: pd.DataFrame, filename: str) -> Path:180 try:181 import pyarrow # noqa182 output_path = self.output_dir / f"{filename}.parquet"183 df.to_parquet(output_path, index=False, engine='pyarrow')184 self.exported_files.append(output_path)185 return output_path186 except ImportError:187 # Fallback to CSV if pyarrow not installed188 return self.export_to_csv(df, filename + "_parquet_fallback")189 190 def export_to_json(self, data: Any, filename: str) -> Path:191 output_path = self.output_dir / f"{filename}.json"192 with open(output_path, 'w', encoding='utf-8') as f:193 json.dump(data, f, indent=2, default=str)194 self.exported_files.append(output_path)195 return output_path196 197 def create_data_model(198 self,199 tables: Dict[str, pd.DataFrame],200 relationships: Optional[List[Dict[str, str]]] = None201 ) -> Dict[str, Any]:202 data_model: Dict[str, Any] = {203 "tables": {},204 "relationships": relationships or [],205 "created_at": datetime.now().isoformat(),206 }207 for table_name, df in tables.items():208 data_model["tables"][table_name] = {209 "columns": df.columns.tolist(),210 "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},211 "row_count": len(df),212 "primary_key": df.columns[0] if len(df.columns) > 0 else None,213 }214 self.export_to_json(data_model, "powerbi_data_model")215 return data_model216 217 def create_analysis_results(218 self,219 ml_results: Dict[str, Any],220 dl_results: Dict[str, Any],221 data_summary: Dict[str, Any],222 ) -> pd.DataFrame:223 rows = [224 {225 "metric_category": "Machine Learning",226 "metric_name": "accuracy" if "accuracy" in ml_results else "mse",227 "metric_value": ml_results.get("accuracy", ml_results.get("mse", 0)),228 "timestamp": datetime.now(),229 },230 {231 "metric_category": "Deep Learning",232 "metric_name": "device",233 "metric_value": str(dl_results.get("device", "unknown")),234 "timestamp": datetime.now(),235 },236 {237 "metric_category": "Data Summary",238 "metric_name": "row_count",239 "metric_value": data_summary.get("row_count", 0),240 "timestamp": datetime.now(),241 },242 ]243 return pd.DataFrame(rows)244 245 def export_predictions(246 self,247 df: pd.DataFrame,248 predictions: List[Any],249 probabilities: Optional[List[List[float]]] = None,250 filename: str = "predictions",251 ) -> Path:252 result_df = df.copy()253 result_df["prediction"] = predictions254 if probabilities is not None:255 prob_array = list(zip(*probabilities))256 for i, probs in enumerate(prob_array):257 result_df[f"prob_class_{i}"] = probs258 return self.export_to_csv(result_df, filename)259 260 def create_dashboard_data(261 self, analysis_results: Dict[str, Any]262 ) -> Dict[str, pd.DataFrame]:263 dashboard_data: Dict[str, pd.DataFrame] = {}264 if "feature_importance" in analysis_results:265 dashboard_data["feature_importance"] = pd.DataFrame(266 analysis_results["feature_importance"]267 )268 if "predictions" in analysis_results:269 dashboard_data["predictions"] = pd.DataFrame(270 analysis_results["predictions"]271 )272 if "metrics" in analysis_results:273 metrics_list = [274 {"metric": k, "value": v}275 for k, v in analysis_results["metrics"].items()276 if isinstance(v, (int, float))277 ]278 if metrics_list:279 dashboard_data["metrics_summary"] = pd.DataFrame(metrics_list)280 return dashboard_data281 282 def export_all(283 self,284 dataframes: Dict[str, pd.DataFrame],285 include_parquet: bool = True,286 ) -> List[Path]:287 exported: List[Path] = []288 for name, df in dataframes.items():289 exported.append(self.export_to_csv(df, name))290 if include_parquet:291 exported.append(self.export_to_parquet(df, name))292 return exported293 294 def get_exported_files(self) -> List[Path]:295 return self.exported_files296 297 def generate_powerbi_instructions(self) -> str:298 return f"""299Power BI Integration Instructions300===================================301 3021. Open Power BI Desktop303 3042. Get Data:305 - Click "Get Data" → "More..."306 - Select "Text/CSV" for CSV files307 - Select "Parquet" for Parquet files308 3093. Load the exported data:310 - Navigate to: {self.output_dir}311 - Select the relevant CSV/Parquet files312 3134. Create relationships (Model view):314 - Drag shared columns between tables to link them315 3165. Build visualizations:317 - Use the "Visualizations" pane to create charts, KPIs, tables318 319Exported files location: {self.output_dir}320Total files exported: {len(self.exported_files)}321"""