import sys import os import argparse import json from pathlib import Path from time import perf_counter import h5py import numpy as np import pandas as pd from sklearn.metrics import mean_squared_error, r2_score from sklearn.model_selection import train_test_split # Add the parent directory to sys.path to import model.py sys.path.append(str(Path(__file__).resolve().parent.parent)) try: from model import Model except ModuleNotFoundError as exc: raise ModuleNotFoundError( "No se pudo importar model.py. Verifica que el archivo exista en la carpeta Solution." ) from exc def _read_csv(path: Path) -> np.ndarray: return pd.read_csv(path, header=None).to_numpy() def _find_data_dir(base_dir: Path) -> Path: candidates = [ base_dir, base_dir / "input_data", base_dir / "dev_phase" / "input_data", base_dir / "final_phase" / "input_data", base_dir / "data", ] required = ("train.csv", "train_labels.csv") for candidate in candidates: if all((candidate / f).exists() for f in required): return candidate raise FileNotFoundError( f"No se encontraron train.csv y train_labels.csv. Revisa la ruta base: {base_dir}" ) def _resolve_test_labels_path(data_dir: Path) -> Path: in_data = data_dir / "test_labels.csv" if in_data.exists(): return in_data if data_dir.name == "input_data": sibling = data_dir.parent / "reference_data" / "test_labels.csv" if sibling.exists(): return sibling return in_data def _load_from_h5(h5_path: Path): with h5py.File(h5_path, "r") as h5f: X_train = h5f["X_train"][:] y_train = h5f["y_train"][:].ravel() X_adapt = h5f["X_adapt"][:] if "X_adapt" in h5f else np.empty((0, X_train.shape[1])) X_test = h5f["X_test"][:] if "X_test" in h5f else np.empty((0, X_train.shape[1])) y_test = h5f["y_test"][:].ravel() if "y_test" in h5f else np.empty((0,)) return X_train, y_train, X_adapt, X_test, y_test def main(): parser = argparse.ArgumentParser( description="Evalua localmente model.py y reporta MSE, R2 y tiempos." ) # Default is picking up data from the parent project directory default_data_dir = str(Path(__file__).resolve().parent.parent.parent / "data" / "dev_phase" / "input_data") parser.add_argument("--h5-file", type=str, default="", help="Ruta a archivo H5 preprocesado.") parser.add_argument("--input-dir", type=str, default=default_data_dir, help="Ruta base de datos.") parser.add_argument("--train-csv", type=str, default="") parser.add_argument("--train-labels-csv", type=str, default="") parser.add_argument("--adapt-csv", type=str, default="") parser.add_argument("--test-csv", type=str, default="") parser.add_argument("--test-labels-csv", type=str, default="") parser.add_argument("--val-size", type=float, default=0.2) parser.add_argument("--seed", type=int, default=42) parser.add_argument("--output-json", type=str, default="") parser.add_argument("--max-fit-samples", type=int, default=0) parser.add_argument("--max-eval-samples", type=int, default=0) args = parser.parse_args() if args.h5_file: h5_path = Path(args.h5_file) X_train_full, y_train_full, X_adapt, X_test, y_test = _load_from_h5(h5_path) path_info = {"h5_file": str(h5_path.resolve())} has_scored_test = X_test.shape[0] > 0 and y_test.shape[0] > 0 else: if args.train_csv and args.train_labels_csv: train_path = Path(args.train_csv) train_labels_path = Path(args.train_labels_csv) adapt_path = Path(args.adapt_csv) if args.adapt_csv else None test_path = Path(args.test_csv) if args.test_csv else None test_labels_path = Path(args.test_labels_csv) if args.test_labels_csv else None else: try: data_dir = _find_data_dir(Path(args.input_dir)) except FileNotFoundError: print(f"Warning: Default data directory not found at {args.input_dir}. Using alternative defaults.") data_dir = _find_data_dir(Path(args.input_dir).parent.parent) # Fall back up if needed train_path = data_dir / "train.csv" train_labels_path = data_dir / "train_labels.csv" adapt_path = data_dir / "train_DA.csv" test_path = data_dir / "test.csv" test_labels_path = _resolve_test_labels_path(data_dir) X_train_full = _read_csv(train_path) y_train_full = _read_csv(train_labels_path).ravel() X_adapt = ( _read_csv(adapt_path) if adapt_path is not None and adapt_path.exists() else np.empty((0, X_train_full.shape[1]), dtype=X_train_full.dtype) ) X_test = ( _read_csv(test_path) if test_path is not None and test_path.exists() else np.empty((0, X_train_full.shape[1]), dtype=X_train_full.dtype) ) y_test = ( _read_csv(test_labels_path).ravel() if test_labels_path is not None and test_labels_path.exists() else np.empty((0,), dtype=y_train_full.dtype) ) path_info = { "train_csv": str(train_path), "train_labels_csv": str(train_labels_path), "adapt_csv": str(adapt_path) if adapt_path is not None else "", "test_csv": str(test_path) if test_path is not None else "", "test_labels_csv": str(test_labels_path) if test_labels_path is not None else "", } has_scored_test = X_test.shape[0] > 0 and y_test.shape[0] > 0 print("Rutas usadas:") print(json.dumps(path_info, indent=2)) if has_scored_test: X_fit, y_fit = X_train_full, y_train_full X_eval, y_eval = X_test, y_test split_name = "test_set" else: X_fit, X_eval, y_fit, y_eval = train_test_split( X_train_full, y_train_full, test_size=args.val_size, random_state=args.seed, ) split_name = "holdout_validation" if args.max_fit_samples and args.max_fit_samples < X_fit.shape[0]: rng = np.random.default_rng(args.seed) idx = rng.choice(X_fit.shape[0], size=args.max_fit_samples, replace=False) X_fit, y_fit = X_fit[idx], y_fit[idx] if args.max_eval_samples and args.max_eval_samples < X_eval.shape[0]: rng = np.random.default_rng(args.seed + 1) idx = rng.choice(X_eval.shape[0], size=args.max_eval_samples, replace=False) X_eval, y_eval = X_eval[idx], y_eval[idx] model = Model() t0 = perf_counter() model.fit(X_fit, y_fit, X_adapt) t1 = perf_counter() y_pred = model.predict(X_eval) t2 = perf_counter() mse = mean_squared_error(y_eval, y_pred) r2 = r2_score(y_eval, y_pred) fit_time = t1 - t0 predict_time = t2 - t1 execution_time = t2 - t0 results = { "split": split_name, "source": "h5" if args.h5_file else "csv", "n_fit": int(X_fit.shape[0]), "n_eval": int(X_eval.shape[0]), "n_adapt": int(X_adapt.shape[0]), "MSE": float(mse), "R2": float(r2), "fit_time_sec": float(fit_time), "predict_time_sec": float(predict_time), "execution_time_sec": float(execution_time), } print(json.dumps(results, indent=2)) if args.output_json: out_path = Path(args.output_json) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(json.dumps(results, indent=2), encoding="utf-8") if __name__ == "__main__": main()