diff --git a/docs/test-scenarios/12-angular-test-cv022-wit230.json b/docs/test-scenarios/12-angular-test-cv022-wit230.json new file mode 100644 index 0000000..2c74fda --- /dev/null +++ b/docs/test-scenarios/12-angular-test-cv022-wit230.json @@ -0,0 +1,29 @@ +{ + "_description": "CenĂ¡rio angular-test-01: CV022 WIT230 com lag e intervalo de datas", + "experimentName": "angular-test-01", + "username": "lucas.kou@aignosi.com.br", + "modelName": "Linear Regression", + "targetVariable": "03CV022/CORRENTE_N_M1_PV(Value)", + "variableColumns": ["303-WIT-230(Value)"], + "lagTrain": {"303-WIT-230(Value)": 3}, + "lagVal": {"303-WIT-230(Value)": 0}, + "remStaticWin": false, + "lowLim": {}, + "uppLim": {}, + "window": 0, + "useScaler": false, + "includeAr": false, + "trainSize": 80, + "shuffle": true, + "lineSeparator": ",", + "decimalSeparator": ".", + "removedIntervals": [], + "degree": 1, + "interactionOnly": false, + "nanTreatment": "drop", + "startDate": "01/05/2022", + "endDate": "31/07/2022", + "scalerName": "None", + "supportFilters": {}, + "staticThreshold": null +} diff --git a/scripts/run_training_test.py b/scripts/run_training_test.py index 2a93c7b..427a8bf 100644 --- a/scripts/run_training_test.py +++ b/scripts/run_training_test.py @@ -1,12 +1,19 @@ #!/usr/bin/env python3 """Utility script to trigger the training workflow end-to-end for testing. -Steps performed: +Steps performed (default): 1. Upload the CSV test dataset to MinIO using the configured `mc` alias. 2. Insert a new experiment_run record in Postgres and capture the generated ID. 3. Trigger the Temporal `train_model` workflow with the correct payload. -Prerequisites: +Local diagnosis (--local / --validate-only): +- --validate-only: Validates scenario parameters only (no MinIO, Postgres, Temporal). +- --local: Runs the same training pipeline locally (validate + load CSV + train + + after_train_calculation). Use to get full Python tracebacks for debugging. + Does not upload to MinIO, insert DB, or start Temporal. By default skips + MLflow save; use --local-save-mlflow to also test saving to MLflow. + +Prerequisites (default flow): - `mc` CLI configured with alias defined in MINIO_ALIAS. - PostgreSQL accessible with credentials in environment variables or defaults. - Temporal server reachable without TLS on TEMPORAL_HOST / TEMPORAL_NAMESPACE. @@ -23,6 +30,7 @@ import subprocess import sys import uuid from datetime import datetime, timedelta +from io import BytesIO from pathlib import Path import psycopg2 @@ -220,6 +228,113 @@ async def trigger_temporal_workflow(workflow_input: dict) -> str: return workflow_id +def _build_local_payload(request_data: dict, csv_path: Path) -> dict: + """Build workflow payload for local run (no real experiment_run_id).""" + return build_workflow_payload( + experiment_run_id=0, + file_name=csv_path.name, + request_data=request_data, + ) + + +def run_validate_only(scenario_name: str) -> dict: + """Validate scenario parameters only. No MinIO, Postgres, or Temporal. + + Returns: + dict: {'success': bool, 'error': str | None, 'scenario': str} + """ + from model_manager.utils.models.train_model_params import TrainModelParams + + result = {'scenario': scenario_name, 'success': False, 'error': None} + try: + request_data = load_scenario(scenario_name) + except FileNotFoundError as exc: + result['error'] = str(exc) + return result + + payload = _build_local_payload(request_data, Path('local.csv')) + try: + train_params = TrainModelParams.from_dict(payload) + train_params.validate_business_rules() + result['success'] = True + except (ValueError, TypeError, KeyError) as e: + result['error'] = str(e) + return result + + +def run_local_pipeline( + scenario_name: str, + csv_path: Path, + save_mlflow: bool = False, +) -> dict: + """Run the same training pipeline locally (validate + train + after_train). + + Reads CSV from disk, runs TrainingRepository.train and after_train_calculation. + Optionally saves to MLflow if save_mlflow is True (requires MLflow env). + + Returns: + dict: {'success': bool, 'error': str | None, 'scenario': str, ...} + """ + from model_manager.utils.logger_helper import get_logger + from model_manager.utils.models.train_model_params import TrainModelParams + from model_manager.utils.repository.training_repository import TrainingRepository + + result = { + 'scenario': scenario_name, + 'success': False, + 'error': None, + } + try: + request_data = load_scenario(scenario_name) + except FileNotFoundError as exc: + result['error'] = str(exc) + return result + + if not csv_path.is_absolute(): + csv_path = PROJECT_ROOT / csv_path + _ensure_source_file(csv_path) + payload = _build_local_payload(request_data, csv_path) + try: + train_params = TrainModelParams.from_dict(payload) + train_params.validate_business_rules() + except (ValueError, TypeError, KeyError) as e: + result['error'] = f'Validation failed: {e}' + return result + + logger = get_logger(__name__) + training_repository = TrainingRepository(logger) + + with open(csv_path, 'rb') as f: + file_content = BytesIO(f.read()) + + try: + train_result = training_repository.train(file_content, train_params) + train_result = training_repository.after_train_calculation( + train_params, train_result + ) + except Exception as e: + result['error'] = str(e) + raise # re-raise so caller gets full traceback for diagnosis + + if save_mlflow: + from model_manager.utils.connectors_config import build_mlflow_config + from model_manager.utils.repository.model_repository import ModelRepository + + mlflow_config = build_mlflow_config() + model_repository = ModelRepository( + url=mlflow_config['url'], + username=mlflow_config['username'], + password=mlflow_config['password'], + logger=logger, + ) + train_result = model_repository.save_model(train_result) + + result['success'] = True + result['run_name'] = getattr(train_result, 'run_name', None) + result['run_dir'] = getattr(train_result, 'run_dir', None) + return result + + def parse_args() -> argparse.Namespace: """Parse command line arguments.""" parser = argparse.ArgumentParser( @@ -241,6 +356,15 @@ Examples: # Run all scenarios python scripts/run_training_test.py --all + + # Validate scenario parameters only (no external services) + python scripts/run_training_test.py --scenario linear-regression-basic --validate-only + + # Run training pipeline locally to diagnose errors (full traceback) + python scripts/run_training_test.py --scenario linear-regression-basic --local --csv docs/test-model-data.csv + + # Local run and save to MLflow (requires MLflow env) + python scripts/run_training_test.py --scenario linear-regression-basic --local --local-save-mlflow """, ) parser.add_argument( @@ -268,6 +392,21 @@ Examples: action='store_true', help='Run all available test scenarios sequentially.', ) + parser.add_argument( + '--validate-only', + action='store_true', + help='Only validate scenario parameters (no MinIO, Postgres, Temporal).', + ) + parser.add_argument( + '--local', + action='store_true', + help='Run training pipeline locally (validate + train from CSV) to get full tracebacks.', + ) + parser.add_argument( + '--local-save-mlflow', + action='store_true', + help='With --local, also save the model to MLflow (requires MLflow env).', + ) return parser.parse_args() @@ -405,12 +544,47 @@ def main() -> None: failed_count = sum(1 for r in results if not r['success']) sys.exit(1 if failed_count > 0 else 0) + # Validate-only mode: no external services + if args.validate_only: + if not args.scenario: + print('Error: --scenario is required with --validate-only.', file=sys.stderr) + sys.exit(1) + result = run_validate_only(args.scenario) + if result['success']: + print(f"Validation OK: {result['scenario']}") + else: + print(f"Validation failed: {result['error']}", file=sys.stderr) + sys.exit(1) + return + + # Local pipeline: same code path as worker, full traceback on error + if args.local: + if not args.scenario: + print('Error: --scenario is required with --local.', file=sys.stderr) + sys.exit(1) + print(f'Running local pipeline: {args.scenario} (CSV: {args.csv})') + result = run_local_pipeline( + args.scenario, + args.csv, + save_mlflow=args.local_save_mlflow, + ) + if not result['success']: + print(f"Error: {result['error']}", file=sys.stderr) + sys.exit(1) + out = {'scenario': result['scenario'], 'success': True} + if result.get('run_name') is not None: + out['run_name'] = result['run_name'] + if result.get('run_dir') is not None: + out['run_dir'] = result['run_dir'] + print(json.dumps(out, indent=2)) + return + # Require scenario argument if not listing or running all if not args.scenario: print('Error: --scenario or --all is required. Use --list to see available scenarios.', file=sys.stderr) sys.exit(1) - # Run single scenario + # Run single scenario (MinIO + Postgres + Temporal) print(f'Running scenario: {args.scenario}') result = run_single_scenario(args.scenario, args.csv)