From 445fe643fee9d581fc212459364b13c6c18a8c9f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 26 May 2026 09:38:24 -0300 Subject: [PATCH] chore: update .gitignore and requirements for development - Added new entries to .gitignore to exclude temporary files and training input datasets, ensuring a cleaner repository. - Included type stubs for psycopg2 in requirements-dev.txt to enhance type checking support for database interactions. - Refactored the run_training_test.py script to implement a structured approach for loading and validating training input JSON files, improving the robustness of the training workflow. --- .gitignore | 9 +- requirements-dev.txt | 1 + scripts/inputs/linear_regression.json | 48 ++++ scripts/inputs/xgboost.json | 63 +++++ scripts/run_training_test.py | 353 ++++++++++++++++++-------- 5 files changed, 361 insertions(+), 113 deletions(-) create mode 100644 scripts/inputs/linear_regression.json create mode 100644 scripts/inputs/xgboost.json diff --git a/.gitignore b/.gitignore index 7b74321..2b0edd4 100644 --- a/.gitignore +++ b/.gitignore @@ -247,4 +247,11 @@ sientia-module/ .secrets .event.json -models/ \ No newline at end of file +models/ +.cursor +openspec + +# Training smoke test: track JSON inputs, ignore local sample CSVs +input_dataset.csv +scripts/**/*.csv +!scripts/inputs/ \ No newline at end of file diff --git a/requirements-dev.txt b/requirements-dev.txt index d1a62ab..e98e671 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -7,6 +7,7 @@ ruff>=0.1.0 # Fast Python linter and formatter (replaces flake8, bl mypy>=1.7.0 # Static type checker bandit>=1.7.5 # Security vulnerability scanner pandas-stubs>=2.0.0 # Type stubs for pandas +types-psycopg2>=2.9.0 # Type stubs for psycopg2 types-requests>=2.31.0 # Type stubs for requests # Testing diff --git a/scripts/inputs/linear_regression.json b/scripts/inputs/linear_regression.json new file mode 100644 index 0000000..b0abdf7 --- /dev/null +++ b/scripts/inputs/linear_regression.json @@ -0,0 +1,48 @@ +{ + "experiment": { + "experiment_run_id": 1001, + "experiment_name": "test-experiment-name", + "run_name": "test-run-name", + "username": "vitor.santos@aignosi.com.br", + "status": "ORCHESTRATOR_WAITING_PROC" + }, + "minio": { + "mc_alias": "suse", + "bucket_name": "model-training", + "file_name": "training-sample-dataset-1001.csv", + "local_csv": "input_dataset.csv" + }, + "temporal": { + "task_queue": "train_model-basic-queue", + "workflow_name": "train_model", + "execution_timeout_minutes": 5, + "run_timeout_minutes": 5, + "task_timeout_minutes": 5 + }, + "payload": { + "experiment_run_id": 1001, + "variable_columns": ["Counter", "Rollout"], + "target_variable": "Square", + "line_separator": ",", + "decimal_separator": ".", + "train_size": 80, + "shuffle": true, + "random_state": 42, + "model_name": "smoke-linreg", + "model_type": "linear_regression", + "model_id": 1001, + "data_model_kwargs": {}, + "model_kwargs": {}, + "opt_params": {}, + "date_column": "timestamp" + }, + "db_only": { + "model_metadata": { + "schemas": { + "components": { + "schemas": {} + } + } + } + } +} diff --git a/scripts/inputs/xgboost.json b/scripts/inputs/xgboost.json new file mode 100644 index 0000000..c1bf718 --- /dev/null +++ b/scripts/inputs/xgboost.json @@ -0,0 +1,63 @@ +{ + "experiment": { + "experiment_run_id": 1002, + "experiment_name": "test-experiment-xgboost", + "run_name": "test-run-xgboost", + "username": "vitor.santos@aignosi.com.br", + "status": "ORCHESTRATOR_WAITING_PROC" + }, + "minio": { + "mc_alias": "suse", + "bucket_name": "model-training", + "file_name": "training-sample-dataset-1002.csv", + "local_csv": "input_dataset.csv" + }, + "temporal": { + "task_queue": "train_model-basic-queue", + "workflow_name": "train_model", + "execution_timeout_minutes": 5, + "run_timeout_minutes": 5, + "task_timeout_minutes": 5 + }, + "payload": { + "experiment_run_id": 1002, + "variable_columns": ["Counter", "Rollout"], + "target_variable": "Square", + "line_separator": ",", + "decimal_separator": ".", + "train_size": 80, + "shuffle": true, + "random_state": 42, + "model_name": "smoke-xgb", + "model_type": "xgboost", + "model_id": 1002, + "data_model_kwargs": { + "scaler_method": "MinMax", + "window_size": 3, + "use_filtering": false, + "transform_mode": "all" + }, + "model_kwargs": {}, + "opt_params": { + "tree_method": "hist", + "device": "cuda", + "learning_rate": 0.3, + "n_estimators": 100, + "max_depth": 32, + "subsample": 0.8, + "colsample_bytree": 0.8, + "min_child_weight": 5, + "random_state": 42 + }, + "date_column": "timestamp" + }, + "db_only": { + "model_metadata": { + "schemas": { + "components": { + "schemas": {} + } + } + } + } +} diff --git a/scripts/run_training_test.py b/scripts/run_training_test.py index 3aaebd4..09f7195 100644 --- a/scripts/run_training_test.py +++ b/scripts/run_training_test.py @@ -17,160 +17,289 @@ # Run cells top to bottom in VS Code / Cursor (**Run Cell** on each `# %%` block). # # Steps mirror `input-sample.md`: optional DB delete + insert, `mc cp` to MinIO, Temporal `train_model`. -# Set `POSTGRES_*`, `TEMPORAL_*`, `TRAIN_TASK_QUEUE`, and configure the `mc` alias (default `suse`). +# Set `POSTGRES_*`, `TEMPORAL_*`, and configure the `mc` alias. Add/remove entries in +# `TRAINING_TEST_INPUTS` below to choose which experiments run. # %% from __future__ import annotations +import csv +import json import os import subprocess import uuid from datetime import datetime, timedelta from pathlib import Path +from typing import Any import psycopg2 from dotenv import load_dotenv from psycopg2.extras import Json from temporalio import client -from dotenv import load_dotenv -load_dotenv() +TRAINING_TEST_INPUTS: list[str] = [ + 'scripts/inputs/linear_regression.json', + 'scripts/inputs/xgboost.json', +] + +_REQUIRED_INPUT_KEYS = ('experiment', 'minio', 'temporal', 'payload', 'db_only') + + +def _load_input(path: Path) -> dict[str, Any]: + """ + Load a training smoke-test input JSON and validate required top-level keys. + + Args: + - path: Path to the input JSON file + + Return: + Parsed input dict with experiment, minio, temporal, payload, and db_only sections + """ + data: dict[str, Any] = json.loads(path.read_text(encoding='utf-8')) + missing = [key for key in _REQUIRED_INPUT_KEYS if key not in data] + if missing: + raise KeyError(f'Input JSON missing required top-level keys: {", ".join(missing)}') + return data + + +def _postgres_connect_kwargs() -> dict[str, str | int]: + """ + Build psycopg2.connect keyword arguments from POSTGRES_* environment variables. + + Return: + host, port, user, password, and dbname suitable for psycopg2.connect + """ + host = os.getenv('POSTGRES_HOST') + user = os.getenv('POSTGRES_USER') + password = os.getenv('POSTGRES_PASSWORD') + dbname = os.getenv('POSTGRES_DBNAME') + if not host or not user or not password or not dbname: + raise RuntimeError( + 'Set POSTGRES_HOST, POSTGRES_USER, POSTGRES_PASSWORD, and POSTGRES_DBNAME' + ) + return { + 'host': host, + 'port': int(os.getenv('POSTGRES_PORT', '5432')), + 'user': user, + 'password': password, + 'dbname': dbname, + } + + +def _resolve_input_paths(project_root: Path, entries: list[str]) -> list[Path]: + """ + Resolve and validate the hardcoded TRAINING_TEST_INPUTS list against the filesystem. + + Args: + - project_root: Repository root used to resolve relative entries + - entries: Input file paths (absolute or relative to project_root) + + Return: + Absolute paths to the input JSON files, in declaration order + """ + if not entries: + raise RuntimeError('TRAINING_TEST_INPUTS is empty; add at least one input JSON path') + resolved: list[Path] = [] + for entry in entries: + candidate = Path(entry) + if not candidate.is_absolute(): + candidate = project_root / candidate + if not candidate.is_file(): + raise FileNotFoundError(f'Training test input file not found: {candidate}') + resolved.append(candidate.resolve()) + return resolved + + +def _validate_csv_columns( + local_csv: Path, + payload: dict[str, Any], + input_path: Path, +) -> None: + """ + Fail fast when the local CSV header does not match payload column names. + + Args: + - local_csv: Resolved path to the CSV on disk + - payload: Training payload with variable_columns, target_variable, and separators + - input_path: Input JSON path (for error messages) + + Return: + None; raises ValueError when required columns are missing from the CSV header + """ + line_separator = str(payload.get('line_separator', ',')) + with local_csv.open(newline='', encoding='utf-8') as handle: + header = next(csv.reader(handle, delimiter=line_separator)) + required = list(payload['variable_columns']) + [str(payload['target_variable'])] + date_column = payload.get('date_column') + if date_column: + required.append(str(date_column)) + missing = [column for column in required if column not in header] + if missing: + raise ValueError( + f'{input_path}: CSV {local_csv} header {header!r} is missing columns ' + f'{missing!r} (check payload variable_columns, target_variable, date_column)' + ) + + +def _resolve_local_csv(project_root: Path, local_csv: str) -> Path: + """ + Resolve a minio.local_csv entry against the project root when relative. + + Args: + - project_root: Repository root + - local_csv: Path declared in the input JSON (absolute or relative) + + Return: + Absolute path to the local CSV file + """ + candidate = Path(local_csv) + if not candidate.is_absolute(): + candidate = project_root / candidate + return candidate + -# %% -# --- configuration (edit here or use `.env` at repo root) --- PROJECT_ROOT = Path(__file__).resolve().parent.parent load_dotenv(PROJECT_ROOT / '.env') -EXPERIMENT_RUN_ID = 1001 -MINIO_MC_ALIAS = os.getenv('MINIO_MC_ALIAS', 'suse') -MINIO_BUCKET = os.getenv('MINIO_DEFAULT_BUCKET', 'model-training') -OBJECT_NAME = f'training-sample-dataset-{EXPERIMENT_RUN_ID}.csv' -LOCAL_CSV = PROJECT_ROOT / 'input_dataset.csv' +experiments: list[dict[str, Any]] = [] +for _input_path in _resolve_input_paths(PROJECT_ROOT, TRAINING_TEST_INPUTS): + _input = _load_input(_input_path) + _local_csv = _resolve_local_csv(PROJECT_ROOT, _input['minio']['local_csv']) + _validate_csv_columns(_local_csv, _input['payload'], _input_path) + experiments.append({'path': _input_path, **_input}) + _payload = _input['payload'] + _experiment = _input['experiment'] + print( + f'input={_input_path} ' + f'model_type={_payload["model_type"]} ' + f'model_name={_payload["model_name"]} ' + f'experiment_run_id={_experiment["experiment_run_id"]}' + ) -PG = { - 'host': os.getenv('POSTGRES_HOST'), - 'port': int(os.getenv('POSTGRES_PORT', '5432')), - 'user': os.getenv('POSTGRES_USER'), - 'password': os.getenv('POSTGRES_PASSWORD'), - 'dbname': os.getenv('POSTGRES_DBNAME'), -} +# %% +# --- configuration (`.env` at repo root; per-run values from input JSON) --- + +PG = _postgres_connect_kwargs() TEMPORAL_HOST = os.getenv('TEMPORAL_HOST') TEMPORAL_NAMESPACE = os.getenv('TEMPORAL_NAMESPACE') -TRAIN_TASK_QUEUE = "train_model-basic-queue" TEMPORAL_TLS = os.getenv('TEMPORAL_USE_TLS', 'false').lower() in ('1', 'true', 'yes') - -print(MINIO_MC_ALIAS, MINIO_BUCKET, OBJECT_NAME, LOCAL_CSV) print(PG) -print(TEMPORAL_HOST, TEMPORAL_NAMESPACE, TRAIN_TASK_QUEUE, TEMPORAL_TLS) +print(TEMPORAL_HOST, TEMPORAL_NAMESPACE, TEMPORAL_TLS) +for exp in experiments: + exp_minio = exp['minio'] + exp_temporal = exp['temporal'] + print( + exp_minio['mc_alias'], + exp_minio['bucket_name'], + exp_minio['file_name'], + _resolve_local_csv(PROJECT_ROOT, exp_minio['local_csv']), + exp_temporal['task_queue'], + ) # %% -# --- 1) database: delete previous row (same id), then insert `experiment_run` --- -# Primary key column is `id` (see `experiment_tracking` updates). `request_data` matches the SQL sample in `input-sample.md`. +# --- 1) database: delete previous row (same id), then insert `experiment_run` for each experiment --- +# Primary key column is `id` (see `experiment_tracking` updates). now = datetime.utcnow() -request_data = { - 'experiment_run_id': EXPERIMENT_RUN_ID, - 'variable_columns': ['feature_a', 'feature_b'], - 'target_variable': 'target', - 'bucket_name': MINIO_BUCKET, - 'file_name': OBJECT_NAME, - 'line_separator': ',', - 'decimal_separator': '.', - 'train_size': 80, - 'shuffle': True, - 'random_state': 42, - 'model_name': 'test-runtime-linear-regression-model', - 'model_type': 'linear_regression', - 'model_id': EXPERIMENT_RUN_ID, - 'data_model_kwargs': {}, - 'model_kwargs': {}, - 'opt_params': {}, - 'date_column': 'timestamp', - 'model_metadata': {'schemas': {'components': {'schemas': {}}}}, -} -with psycopg2.connect(**PG) as conn: +with psycopg2.connect( + host=PG['host'], + port=PG['port'], + user=PG['user'], + password=PG['password'], + dbname=PG['dbname'], +) as conn: with conn.cursor() as cur: - cur.execute('DELETE FROM public.experiment_run WHERE id = %s', (EXPERIMENT_RUN_ID,)) - cur.execute( - """ - INSERT INTO public.experiment_run ( - id, experiment_name, run_name, username, status, error_message, - created_at, updated_at, bucket_name, file_name, request_data, orchestrator_response_data + for exp in experiments: + exp_experiment = exp['experiment'] + exp_minio = exp['minio'] + exp_payload = exp['payload'] + exp_db_only = exp['db_only'] + request_data = { + **exp_payload, + 'bucket_name': exp_minio['bucket_name'], + 'file_name': exp_minio['file_name'], + 'model_metadata': exp_db_only['model_metadata'], + } + cur.execute( + 'DELETE FROM public.experiment_run WHERE id = %s', + (exp_experiment['experiment_run_id'],), + ) + cur.execute( + """ + INSERT INTO public.experiment_run ( + id, experiment_name, run_name, username, status, error_message, + created_at, updated_at, bucket_name, file_name, request_data, orchestrator_response_data + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + """, + ( + exp_experiment['experiment_run_id'], + exp_experiment['experiment_name'], + exp_experiment['run_name'], + exp_experiment['username'], + exp_experiment['status'], + None, + now, + now, + exp_minio['bucket_name'], + exp_minio['file_name'], + Json(request_data), + None, + ), ) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) - """, - ( - EXPERIMENT_RUN_ID, - 'test-experiment-name', - 'test-run-name', - 'vitor.santos@aignosi.com.br', - 'ORCHESTRATOR_WAITING_PROC', - None, - now, - now, - MINIO_BUCKET, - OBJECT_NAME, - Json(request_data), - None, - ), - ) # %% -# --- 2) MinIO: upload local CSV (requires `mc` CLI and alias configured) --- -subprocess.run( - ['mc', 'cp', str(LOCAL_CSV), f'{MINIO_MC_ALIAS}/{MINIO_BUCKET}/{OBJECT_NAME}'], - check=True, -) +# --- 2) MinIO: upload local CSV for each experiment (requires `mc` CLI and alias configured) --- +for exp in experiments: + exp_minio = exp['minio'] + local_csv = _resolve_local_csv(PROJECT_ROOT, exp_minio['local_csv']) + subprocess.run( + [ + 'mc', + 'cp', + str(local_csv), + f'{exp_minio["mc_alias"]}/{exp_minio["bucket_name"]}/{exp_minio["file_name"]}', + ], + check=True, + ) # %% -# --- 3) Temporal: start `train_model` (flat payload; worker fills `model_metadata` in `load_model_metadata`) --- +# --- 3) Temporal: connect once, then start `train_model` per experiment --- -if not TEMPORAL_HOST or not TEMPORAL_NAMESPACE or not TRAIN_TASK_QUEUE: - raise RuntimeError('Set TEMPORAL_HOST, TEMPORAL_NAMESPACE, and TRAIN_TASK_QUEUE') -TH, TN, TQ = TEMPORAL_HOST, TEMPORAL_NAMESPACE, TRAIN_TASK_QUEUE +if not TEMPORAL_HOST or not TEMPORAL_NAMESPACE: + raise RuntimeError('Set TEMPORAL_HOST and TEMPORAL_NAMESPACE in the environment') -_workflow_input = { - 'experiment_run_id': EXPERIMENT_RUN_ID, - 'variable_columns': ['Counter', 'Rollout'], - 'target_variable': 'Square', - 'bucket_name': MINIO_BUCKET, - 'file_name': OBJECT_NAME, - 'line_separator': ',', - 'decimal_separator': '.', - 'train_size': 80, - 'shuffle': True, - 'random_state': 42, - 'model_name': 'test-runtime', - 'model_type': 'linear_regression', - 'model_id': EXPERIMENT_RUN_ID, - 'data_model_kwargs': {}, - 'model_kwargs': {}, - 'opt_params': {}, - 'date_column': 'timestamp', -} - -c = await client.Client.connect( - target_host=TH, - namespace=TN, +c = await client.Client.connect( # type: ignore[top-level-await] + target_host=TEMPORAL_HOST, + namespace=TEMPORAL_NAMESPACE, tls=TEMPORAL_TLS, ) # %% -wid = f'train-model-test-{uuid.uuid4()}' -result = await c.execute_workflow( # type: ignore[call-overload] - 'train_model', - _workflow_input, - id=wid, - task_queue=TQ, - execution_timeout=timedelta(minutes=5), - run_timeout=timedelta(minutes=5), - task_timeout=timedelta(minutes=5), -) -print(wid) -print(result) +for exp in experiments: + exp_minio = exp['minio'] + exp_temporal = exp['temporal'] + exp_payload = exp['payload'] + workflow_input = { + **exp_payload, + 'bucket_name': exp_minio['bucket_name'], + 'file_name': exp_minio['file_name'], + } + wid = f'train-model-test-{uuid.uuid4()}' + result = await c.execute_workflow( # type: ignore[top-level-await, call-overload] + exp_temporal['workflow_name'], + workflow_input, + id=wid, + task_queue=exp_temporal['task_queue'], + execution_timeout=timedelta(minutes=exp_temporal['execution_timeout_minutes']), + run_timeout=timedelta(minutes=exp_temporal['run_timeout_minutes']), + task_timeout=timedelta(minutes=exp_temporal['task_timeout_minutes']), + ) + print(wid) + print(result) -# %% \ No newline at end of file +# %%