Spaces:
Sleeping
Sleeping
File size: 3,628 Bytes
1380c2c fa3d628 1380c2c fa3d628 1380c2c fa3d628 1380c2c fa3d628 1380c2c fa3d628 1380c2c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Point d'entree CLI pour regenerer les artefacts historiques P1."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from scripts.experience_1 import run_experience_1
from scripts.pipeline_utils import ensure_paths_exist, relative_to_project
from scripts.runtime_model_specs import HISTORICAL_RUNTIME_MODEL_SPEC
EXPERIENCE_1_SCRIPT_PATH = Path("scripts/experience_1.py")
HISTORICAL_OUTPUTS = [
Path("artifacts/experiments/experience_1/dataset_consolide_historique_colonnes.csv"),
Path("artifacts/experiments/experience_1/model_results.csv"),
HISTORICAL_RUNTIME_MODEL_SPEC.output_model_path.relative_to(PROJECT_ROOT),
HISTORICAL_RUNTIME_MODEL_SPEC.output_metadata_path.relative_to(PROJECT_ROOT),
]
HISTORICAL_METADATA_PATH = HISTORICAL_RUNTIME_MODEL_SPEC.output_metadata_path.relative_to(PROJECT_ROOT)
def parse_args() -> argparse.Namespace:
"""Construit l'interface en ligne de commande du script."""
parser = argparse.ArgumentParser(
description="Execute experience_1 headlessly and validate the historical model artifacts.",
)
parser.add_argument("--tracking-uri", default=None, help="Optional MLflow tracking URI override.")
parser.add_argument("--cv-splits", type=int, default=4, help="Number of grouped CV folds.")
parser.add_argument("--seed", type=int, default=42, help="Random seed used for the experiment.")
return parser.parse_args()
def train_historical_model(
*,
tracking_uri: str | None = None,
cv_splits: int = 4,
seed: int = 42,
) -> dict[str, object]:
"""Execute `scripts/experience_1.py` et valide les artefacts historiques.
Args:
tracking_uri: Tracking URI MLflow optionnel.
cv_splits: Nombre de folds pour la CV groupee.
seed: Graine aleatoire globale.
Returns:
dict[str, object]: Resume du modele historique et de ses artefacts.
"""
print(f"[historical] Executing {relative_to_project(EXPERIENCE_1_SCRIPT_PATH)}")
run_experience_1(
tracking_uri=tracking_uri,
cv_n_splits=cv_splits,
seed=seed,
)
resolved_outputs = ensure_paths_exist(HISTORICAL_OUTPUTS, label="historical model outputs")
metadata = json.loads((PROJECT_ROOT / HISTORICAL_METADATA_PATH).read_text(encoding="utf-8"))
metrics = metadata.get("metrics", {})
print(
"[historical] Outputs validated "
f"(model={metadata.get('model_name')}, test_rmse={metrics.get('test_rmse')}, test_r2={metrics.get('test_r2')})"
)
return {
"script": relative_to_project(EXPERIENCE_1_SCRIPT_PATH),
"artifact_source": "retrained",
"training_notebook_reference": metadata.get("training_notebook"),
"outputs": [relative_to_project(path) for path in resolved_outputs],
"model_name": metadata.get("model_name"),
"target_year": metadata.get("target_year"),
"registered_model_name": metadata.get("registered_model_name"),
"registered_model_version": metadata.get("registered_model_version"),
"registered_model_run_id": metadata.get("registered_model_run_id"),
"model_uri": metadata.get("model_uri"),
"metrics": metrics,
}
def main() -> None:
"""Execute le script historique depuis la CLI."""
args = parse_args()
train_historical_model(
tracking_uri=args.tracking_uri,
cv_splits=args.cv_splits,
seed=args.seed,
)
if __name__ == "__main__":
main()
|