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

chore: update .gitignore and requirements for development

Browse Source 
- 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.
This commit is contained in:
vitor-aignosi
2026-05-26 09:38:24 -03:00
parent 64295d3855
commit 445fe643fe
5 changed files with 361 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.gitignore vendored
View File

@@ -248,3 +248,10 @@ sientia-module/
.event.json
models/
.cursor
openspec
# Training smoke test: track JSON inputs, ignore local sample CSVs
input_dataset.csv
scripts/**/*.csv
!scripts/inputs/

1
requirements-dev.txt
View File

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

48
scripts/inputs/linear_regression.json Normal file
View File

@@ -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": {}
}
}
}
}
}

63
scripts/inputs/xgboost.json Normal file
View File

@@ -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": {}
}
}
}
}
}

305
scripts/run_training_test.py
View File

@@ -17,83 +17,216 @@
# 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,))
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 (
@@ -103,72 +236,68 @@ with psycopg2.connect(**PG) as conn:
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',
exp_experiment['experiment_run_id'],
exp_experiment['experiment_name'],
exp_experiment['run_name'],
exp_experiment['username'],
exp_experiment['status'],
None,
now,
now,
MINIO_BUCKET,
OBJECT_NAME,
exp_minio['bucket_name'],
exp_minio['file_name'],
Json(request_data),
None,
),
)
# %%
# --- 2) MinIO: upload local CSV (requires `mc` CLI and alias configured) ---
# --- 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'{MINIO_MC_ALIAS}/{MINIO_BUCKET}/{OBJECT_NAME}'],
[
'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,
)
# %%
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[call-overload]
'train_model',
_workflow_input,
result = await c.execute_workflow( # type: ignore[top-level-await, call-overload]
exp_temporal['workflow_name'],
workflow_input,
id=wid,
task_queue=TQ,
execution_timeout=timedelta(minutes=5),
run_timeout=timedelta(minutes=5),
task_timeout=timedelta(minutes=5),
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)