gitea_admin/sientia-dataops-model-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1579: Added local and validation flags to test script

Browse Source
This commit is contained in:
Kou Kinoshita
2026-02-13 17:51:17 -03:00
parent f0cdace820
commit 9e3585afce
2 changed files with 206 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
docs/test-scenarios/12-angular-test-cv022-wit230.json Normal file
View File

@@ -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
}

180
scripts/run_training_test.py
View File

@@ -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)