{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":94147,"databundleVersionId":11390004,"sourceType":"competition"}],"dockerImageVersionId":30918,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import h5py\nimport numpy as np\nimport pandas as pd\nfrom sklearn.pipeline import Pipeline\nfrom sklearn.preprocessing import StandardScaler\nfrom sklearn.multioutput import MultiOutputRegressor\nfrom sklearn.ensemble import RandomForestRegressor\nimport matplotlib.pyplot as plt\n\nclass CellTypePipeline:\n    \"\"\"\n    A pipeline for loading cell type data from an H5 file, training a model to predict \n    cell type abundances based on spot coordinates, and generating a submission file.\n    \"\"\"\n    \n    def __init__(self, h5_file_path):\n        self.h5_file_path = h5_file_path\n        self.train_spot_tables = {}\n        self.cell_type_columns = None\n        \n    def load_train_data(self):\n        \"\"\"\n        Load training spot data from the H5 file and store each slide as a DataFrame.\n        \"\"\"\n        with h5py.File(self.h5_file_path, \"r\") as f:\n            train_spots = f[\"spots/Train\"]\n            for slide_name in train_spots.keys():\n                spot_array = np.array(train_spots[slide_name])\n                df = pd.DataFrame(spot_array)\n                self.train_spot_tables[slide_name] = df\n        print(\"Training data loaded successfully.\")\n        \n    def prepare_training_set(self, slide_id='S_1'):\n        \"\"\"\n        Prepare training features and targets from a given slide.\n        \n        Parameters:\n            slide_id (str): Identifier for the training slide to use.\n        \n        Returns:\n            X (ndarray): Features array (using 'x' and 'y' columns).\n            y (ndarray): Target cell type abundances.\n        \"\"\"\n        if slide_id not in self.train_spot_tables:\n            raise ValueError(f\"Slide {slide_id} not found in training data.\")\n        df = self.train_spot_tables[slide_id]\n        # Assume first two columns are coordinates and the rest are cell type abundances.\n        feature_cols = ['x', 'y']\n        target_cols = [col for col in df.columns if col not in feature_cols]\n        self.cell_type_columns = target_cols\n        X = df[feature_cols].values.astype(float)\n        y = df[target_cols].values.astype(float)\n        return X, y\n    \n    def load_test_data(self, slide_id):\n        \"\"\"\n        Load test spot data for a given slide.\n        \n        Parameters:\n            slide_id (str): Identifier for the test slide.\n        \n        Returns:\n            test_df (DataFrame): DataFrame containing test spot coordinates.\n        \"\"\"\n        with h5py.File(self.h5_file_path, \"r\") as f:\n            test_spots = f[\"spots/Test\"]\n            if slide_id not in test_spots:\n                raise ValueError(f\"Slide {slide_id} not found in test data.\")\n            spot_array = np.array(test_spots[slide_id])\n            test_df = pd.DataFrame(spot_array)\n        print(f\"Test data for slide {slide_id} loaded successfully.\")\n        return test_df\n    \n    def build_model_pipeline(self):\n        \"\"\"\n        Build and return a scikit-learn pipeline that scales data and fits a multi-output regressor.\n        \"\"\"\n        pipeline = Pipeline([\n            ('scaler', StandardScaler()),\n            ('regressor', MultiOutputRegressor(RandomForestRegressor(n_estimators=100, random_state=42)))\n        ])\n        return pipeline\n    \n    def train(self, X, y):\n        \"\"\"\n        Train the model pipeline on provided features and targets.\n        \n        Parameters:\n            X (ndarray): Training features.\n            y (ndarray): Training target values.\n        \n        Returns:\n            model (Pipeline): Trained scikit-learn pipeline.\n        \"\"\"\n        model = self.build_model_pipeline()\n        model.fit(X, y)\n        print(\"Model training complete.\")\n        return model\n    \n    def predict(self, model, X_test):\n        \"\"\"\n        Make predictions on test data using the trained model.\n        \n        Parameters:\n            model (Pipeline): Trained model pipeline.\n            X_test (ndarray): Test features.\n        \n        Returns:\n            predictions (ndarray): Predicted cell type abundances.\n        \"\"\"\n        predictions = model.predict(X_test)\n        return predictions\n    \n    def create_submission(self, test_df, predictions, submission_filename=\"submission.csv\"):\n        \"\"\"\n        Create a submission CSV file with predicted cell type abundances.\n        \n        Parameters:\n            test_df (DataFrame): Original test DataFrame (to get the index as spot IDs).\n            predictions (ndarray): Predicted cell type abundances.\n            submission_filename (str): Name of the CSV file to be created.\n        \"\"\"\n        pred_df = pd.DataFrame(predictions, columns=self.cell_type_columns, index=test_df.index)\n        pred_df.insert(0, 'ID', pred_df.index)\n        pred_df.to_csv(submission_filename, index=False)\n        print(f\"Submission file '{submission_filename}' created!\")\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n# Example usage:\nif __name__ == \"__main__\":\n    # File path to the dataset\n    h5_file_path = \"/kaggle/input/el-hackathon-2025/elucidata_ai_challenge_data.h5\"\n    \n    # Initialize the pipeline\n    pipeline = CellTypePipeline(h5_file_path)\n    \n    # Load training data and prepare training set from a specific slide (e.g., 'S_1')\n    pipeline.load_train_data()\n    X_train, y_train = pipeline.prepare_training_set(slide_id='S_1')\n    \n    # Train the model\n    model = pipeline.train(X_train, y_train)\n    \n    # Load test data for a specific slide (e.g., 'S_7')\n    test_df = pipeline.load_test_data(slide_id='S_7')\n    X_test = test_df[['x', 'y']].values.astype(float)\n    \n    # Predict cell type abundances on test data\n    predictions = pipeline.predict(model, X_test)\n    \n    # Create and save the submission file\n    pipeline.create_submission(test_df, predictions, submission_filename=\"submission.csv\")\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}