#!/usr/bin/env python3 """ Model classes for crop yield prediction """ import pandas as pd import numpy as np from typing import Dict, Any, List class DataPreprocessor: """Data preprocessing class that matches the trained model.""" def __init__(self): self.label_encoders = {} self.scaler = None self.imputer = None self.feature_names = None def prepare_features(self, df): """Prepare features for prediction.""" data = df.copy() # Feature engineering (same as training) data['Area_Production_Ratio'] = data['Area'] / (data['Production'] + 1e-6) data['Yield_Area_Interaction'] = data.get('Yield', 0) * data['Area'] data['Production_Per_Area'] = data['Production'] / (data['Area'] + 1e-6) # Create season dummies season_dummies = pd.get_dummies(data['Season'], prefix='Season') expected_seasons = ['Season_Autumn', 'Season_Kharif', 'Season_Rabi', 'Season_Summer', 'Season_Total', 'Season_Whole Year', 'Season_Winter'] for season in expected_seasons: if season not in season_dummies.columns: season_dummies[season] = 0 data = pd.concat([data, season_dummies[expected_seasons]], axis=1) # Handle categorical variables categorical_cols = ['State', 'District', 'Crop'] for col in categorical_cols: if col in data.columns and col in self.label_encoders: unique_values = set(data[col].astype(str)) known_values = set(self.label_encoders[col].classes_) new_values = unique_values - known_values if new_values: mode_value = self.label_encoders[col].classes_[0] data[col] = data[col].astype(str).replace(list(new_values), mode_value) data[f'{col}_encoded'] = self.label_encoders[col].transform(data[col].astype(str)) elif col in data.columns: data[f'{col}_encoded'] = pd.Categorical(data[col].astype(str)).codes # Select features feature_cols = ['Crop_Year', 'Area', 'Production', 'Annual_Rainfall', 'Fertilizer', 'Pesticide', 'State_encoded', 'Crop_encoded', 'Area_Production_Ratio', 'Yield_Area_Interaction', 'Production_Per_Area'] + expected_seasons if 'District_encoded' in data.columns: feature_cols.append('District_encoded') available_cols = [col for col in feature_cols if col in data.columns] X = data[available_cols].copy() return X, data def transform(self, X): """Transform features using fitted preprocessors.""" if self.imputer is None or self.scaler is None: raise ValueError("Preprocessor not fitted.") X_imputed = pd.DataFrame( self.imputer.transform(X), columns=X.columns, index=X.index ) X_scaled = pd.DataFrame( self.scaler.transform(X_imputed), columns=X.columns, index=X.index ) return X_scaled