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

Merge pull request #20 from Aignosi/SIENTIAPDE-1182-ajustar-laborious-para-pegar-timestamp-da-resposta-do-mlflow

Browse Source 
SIENTIAPDE-1182: Enhance tests and refactor data structures for improved clarity and functionality
This commit is contained in:
Bruno Domingues
2025-09-05 10:39:41 -03:00
committed by GitHub
parent 460bfe4073 f971e711bd
commit 8340189741
6 changed files with 313 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
laborious/activities/gates.py
View File

@@ -41,6 +41,7 @@ mlflow_response_filter_functions = {
# MLFlow content filter function mappings
mlflow_content_filter_functions = {
'NAN_VALUES': nan_values_filter,
'EMPTY_DATA': filter_empty_data,
'path_confidence': {
'STOP': -1,
'CONTINUE': 18,
@@ -403,36 +404,36 @@ class Gates(BaseActivity):
data = DataFrame(input_data['data'])
# Create timestamp column from index and reset index
data['timestamp'] = data.index
data = data.reset_index(drop=True)
self.debug(
f"Prediction store policy: {prediction_store_policy}", metadata)
self.debug(f"Prediction data: {data.to_string()}", metadata)
policy_type, policy_value = self.get_prediction_store_policy(
prediction_store_policy, metadata)
# If data has no timestamp, we use the default timestamp and not sort the data
if 'timestamp' not in data.columns:
self.warning(
"Data has no timestamp, using default timestamp", metadata)
data['timestamp'] = input_data['timestamp']
else:
self.debug(
"Data has timestamp, sorting data by timestamp", metadata)
self.info(
f"Sorting data by timestamp and applying policy: {policy_type}:{policy_value}", metadata)
# If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
if policy_type == 'lts':
self.debug(
"Sorting data by timestamp descending", metadata)
data = data.sort_values(by='timestamp', ascending=False)
# If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
elif policy_type == 'erl':
self.debug(
"Sorting data by timestamp ascending", metadata)
data = data.sort_values(by='timestamp', ascending=True)
else:
self.error(
f"Invalid policy type: {policy_type}, using default policy", metadata)
raise ValueError(
f"Invalid policy type: {policy_type}")
# If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
if policy_type == 'lts':
self.debug(
"Sorting data by timestamp descending", metadata)
data = data.sort_values(by='timestamp', ascending=False)
# If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
elif policy_type == 'erl':
self.debug(
"Sorting data by timestamp ascending", metadata)
data = data.sort_values(by='timestamp', ascending=True)
else:
self.error(
f"Invalid policy type: {policy_type}, using default policy", metadata)
raise ValueError(
f"Invalid policy type: {policy_type}")
data = data.head(int(policy_value))
@@ -443,7 +444,7 @@ class Gates(BaseActivity):
data = data.sort_values(by='timestamp', ascending=False)
data = data.reset_index(drop=True)
self.info(f"Prediction formatted: {data.size} rows", metadata)
self.info(f"Prediction formatted: {len(data)} rows", metadata)
self.debug(f"Prediction data: {data.to_string()}", metadata)
return data.to_dict()

4
laborious/utils/repository/model_repository.py
View File

@@ -39,6 +39,7 @@ class MLFlowRepository():
"""
try:
return {
'success': True,
'content': self.model_serving.get_cached_transform(
@@ -67,12 +68,15 @@ class MLFlowRepository():
- dict: A dictionary containing the predicted data.
"""
try:
input_index = data.index
start_time = datetime.now()
data = self.model_serving.get_cached_predict(
model_name, data, model_retention)
end_time = datetime.now()
data = pd.DataFrame(data, columns=['prediction'])
data.index = input_index
data['response_time'] = (end_time - start_time).total_seconds()
return {

260
tests.ipynb
View File

@@ -2,7 +2,7 @@
"cells": [
{
"cell_type": "code",
"execution_count": null,
"execution_count": 1,
"id": "b10e5c25",
"metadata": {},
"outputs": [
@@ -110,12 +110,12 @@
" 0.06282208 -0.79004057 -0.20289499 0.83467535 0.26166448 -0.64498684\n",
" -0.32228873 -0.61679398 -0.95035374 0.85492092 -0.10358534 -0.38492986\n",
" 0.19695438 -0.98537109 -0.44395579 0.40606693 0.26753955]\n",
" Counter Rollout CounterPlusRollout Timestamp\n",
"0 0.000000 0.000000 0.000000 2025-01-01 00:00:00\n",
"1 0.547912 0.284967 0.832879 2025-01-01 00:00:01\n",
"2 0.425669 0.842959 1.268628 2025-01-01 00:00:02\n",
"3 1.142865 0.112064 1.254928 2025-01-01 00:00:03\n",
"4 1.537601 0.184200 1.721801 2025-01-01 00:00:04\n"
" Counter Rollout CounterPlusRollout Timestamp\n",
"0 37.270658 -59.474345 -22.203688 2025-01-01 00:00:00\n",
"1 43.561734 -54.184658 -10.622924 2025-01-01 00:00:01\n",
"2 42.158150 -43.826926 -1.668776 2025-01-01 00:00:02\n",
"3 50.392926 -57.394165 -7.001239 2025-01-01 00:00:03\n",
"4 54.925249 -56.055140 -1.129892 2025-01-01 00:00:04\n"
]
}
],
@@ -165,32 +165,244 @@
},
{
"cell_type": "code",
"execution_count": 2,
"id": "e7c8eeb1",
"execution_count": 8,
"id": "c61be7ab",
"metadata": {},
"outputs": [
{
"ename": "TypeError",
"evalue": "MLFlowRepository.__init__() missing 1 required positional argument: 'logger'",
"output_type": "error",
"traceback": [
"\u001b[31m---------------------------------------------------------------------------\u001b[39m",
"\u001b[31mTypeError\u001b[39m Traceback (most recent call last)",
"\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[2]\u001b[39m\u001b[32m, line 3\u001b[39m\n\u001b[32m 1\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mlaborious\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mutils\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mrepository\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mmodel_repository\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m MLFlowRepository\n\u001b[32m----> \u001b[39m\u001b[32m3\u001b[39m mlflow_repository = \u001b[43mMLFlowRepository\u001b[49m\u001b[43m(\u001b[49m\n\u001b[32m 4\u001b[39m \u001b[43m \u001b[49m\u001b[43mhost\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mhttp://localhost:5080/\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m 5\u001b[39m \u001b[43m \u001b[49m\u001b[43musername\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\n\u001b[32m 6\u001b[39m \u001b[43m \u001b[49m\u001b[43mpassword\u001b[49m\u001b[43m=\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43maignosi\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\n\u001b[32m 7\u001b[39m \u001b[43m)\u001b[49m\n\u001b[32m 9\u001b[39m mlflow_repository.get_experiment_by_run_id(\u001b[33m\"\u001b[39m\u001b[33m1\u001b[39m\u001b[33m\"\u001b[39m)\n",
"\u001b[31mTypeError\u001b[39m: MLFlowRepository.__init__() missing 1 required positional argument: 'logger'"
"name": "stdout",
"output_type": "stream",
"text": [
"[ 0.5479121 -0.12224312 0.71719584 0.39473606 -0.8116453 0.9512447\n",
" 0.5222794 0.57212861 -0.74377273 -0.09922812 -0.25840395 0.85352998\n",
" 0.28773024 0.64552323 -0.1131716 -0.54552256 0.10916957 -0.87236549\n",
" 0.65526234 0.2633288 0.51617548 -0.29094806 0.94139605 0.78624224\n",
" 0.55676699 -0.61072258 -0.06655799 -0.91239247 -0.69142102 0.36609791\n",
" 0.48952431 0.93501946 -0.34834928 -0.25908059 -0.06088838 -0.62105728\n",
" -0.74015699 -0.04859015 -0.5461813 0.33962799 -0.12569616 0.66535639\n",
" 0.4005302 -0.37526672 0.6645196 0.60952871 -0.22504324 -0.42334379\n",
" 0.36499101]\n",
"[-0.72049503 -0.6001836 -0.98527546 0.57384876 0.32970171 0.41033076\n",
" 0.56145806 -0.08216845 0.13748239 -0.720406 -0.77093985 0.33680592\n",
" -0.05780759 0.13047221 0.52999771 0.26943664 0.1071588 0.11841432\n",
" -0.3920998 -0.93836433 -0.12656522 -0.57083065 -0.18294271 0.70680615\n",
" -0.53212103 -0.88339452 -0.43723222 -0.41281248 0.32383303 0.1140643\n",
" 0.56779642 0.32862708 -0.18722628 0.62804077 -0.66605416 -0.95457585\n",
" -0.81990428 0.4447187 -0.07624554 -0.67745644 0.00208955 -0.69537579\n",
" 0.39264075 -0.10768745 -0.23795755 -0.39697582 0.26056519 -0.27637478\n",
" -0.82470016]\n",
" Counter Rollout Square Timestamp\n",
"0 -100.000000 100.000000 -37.646304 2025-01-01 00:00:00\n",
"1 -79.942159 79.589040 -72.994086 2025-01-01 00:00:05\n",
"2 -84.417207 62.586392 -30.252436 2025-01-01 00:00:10\n",
"3 -58.162265 34.674447 7.555532 2025-01-01 00:00:15\n",
"4 -43.711857 50.931053 26.035410 2025-01-01 00:00:20\n"
]
}
],
"source": [
"from laborious.utils.repository.model_repository import MLFlowRepository\n",
"import pandas as pd\n",
"import numpy as np\n",
"\n",
"mlflow_repository = MLFlowRepository(\n",
" host=\"http://localhost:5080/\",\n",
" username=\"aignosi\",\n",
" password=\"aignosi\"\n",
")\n",
"# Set random seed for reproducibility\n",
"rng = np.random.default_rng(42)\n",
"\n",
"mlflow_repository.get_experiment_by_run_id(\"1\")"
"size = 50\n",
"\n",
"# Generate random walks starting at 0\n",
"counter = np.zeros(size)\n",
"rollout = np.zeros(size)\n",
"square = np.zeros(size)\n",
"\n",
"# Generate random steps between -1 and 1\n",
"counter_steps = rng.uniform(-1, 1, size-1)\n",
"rollout_steps = rng.uniform(-1, 1, size-1)\n",
"square_steps = rng.uniform(-1, 1, size-1)\n",
"\n",
"print(counter_steps)\n",
"print(rollout_steps)\n",
"\n",
"# Calculate cumulative sum and scale to -100 to 100 range\n",
"for i in range(1, size):\n",
" counter[i] = counter[i-1] + counter_steps[i-1]\n",
" rollout[i] = rollout[i-1] + rollout_steps[i-1]\n",
" square[i] = square[i-1] + square_steps[i-1]\n",
"\n",
"# normalize values between -100 and 100, lowest value is -100, highest value is 100\n",
"counter = (counter - min(counter)) / (max(counter) - min(counter)) * 200 - 100\n",
"rollout = (rollout - min(rollout)) / (max(rollout) - min(rollout)) * 200 - 100\n",
"square = (square - min(square)) / (max(square) - min(square)) * 200 - 100\n",
"\n",
"# Create DataFrame\n",
"df = pd.DataFrame({\n",
" 'Counter': counter,\n",
" 'Rollout': rollout,\n",
" 'Square': square\n",
"})\n",
"\n",
"# add a timestamp column\n",
"df['Timestamp'] = pd.date_range(start='2025-01-01', periods=size, freq='5s')\n",
"\n",
"# Save to CSV\n",
"df.to_csv('random_walks_demo.csv', index=False)\n",
"\n",
"# Display first few rows\n",
"print(df.head())\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "e7c8eeb1",
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>a</th>\n",
" <th>b</th>\n",
" </tr>\n",
" <tr>\n",
" <th>timestamp</th>\n",
" <th></th>\n",
" <th></th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>2025-01-01</th>\n",
" <td>1</td>\n",
" <td>4</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2025-01-02</th>\n",
" <td>2</td>\n",
" <td>5</td>\n",
" </tr>\n",
" <tr>\n",
" <th>2025-01-03</th>\n",
" <td>3</td>\n",
" <td>6</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"</div>"
],
"text/plain": [
" a b\n",
"timestamp \n",
"2025-01-01 1 4\n",
"2025-01-02 2 5\n",
"2025-01-03 3 6"
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"True\n"
]
},
{
"data": {
"text/html": [
"<div>\n",
"<style scoped>\n",
" .dataframe tbody tr th:only-of-type {\n",
" vertical-align: middle;\n",
" }\n",
"\n",
" .dataframe tbody tr th {\n",
" vertical-align: top;\n",
" }\n",
"\n",
" .dataframe thead th {\n",
" text-align: right;\n",
" }\n",
"</style>\n",
"<table border=\"1\" class=\"dataframe\">\n",
" <thead>\n",
" <tr style=\"text-align: right;\">\n",
" <th></th>\n",
" <th>a</th>\n",
" <th>b</th>\n",
" </tr>\n",
" <tr>\n",
" <th>timestamp</th>\n",
" <th></th>\n",
" <th></th>\n",
" </tr>\n",
" </thead>\n",
" <tbody>\n",
" <tr>\n",
" <th>2025-01-03</th>\n",
" <td>3</td>\n",
" <td>6</td>\n",
" </tr>\n",
" </tbody>\n",
"</table>\n",
"</div>"
],
"text/plain": [
" a b\n",
"timestamp \n",
"2025-01-03 3 6"
]
},
"metadata": {},
"output_type": "display_data"
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"3\n",
"1\n"
]
}
],
"source": [
"from pandas import DataFrame\n",
"\n",
"a = DataFrame({\n",
" \"a\": {\"2025-01-01\": 1, \"2025-01-02\": 2, \"2025-01-03\": 3},\n",
" \"b\": {\"2025-01-01\": 4, \"2025-01-02\": 5, \"2025-01-03\": 6},\n",
"})\n",
"\n",
"a.index.name = \"timestamp\"\n",
"\n",
"display(a)\n",
"\n",
"print(\"a\" in a.columns)\n",
"\n",
"b = a.tail(1)\n",
"\n",
"display(b)\n",
"\n",
"print(len(a))\n",
"print(len(b))\n",
"print(b.size)\n"
]
}
],

40
tests/laborious/activities/test_gates.py
View File

@@ -379,8 +379,14 @@ async def test_format_prediction_no_timestamp(gates_activity):
# Arrange
input_data = {
**metadata,
'data': {'prediction': [1], 'response_time': [0.1]},
'timestamp': '2023-05-26 11:12:27',
'data': {
'prediction': {
'2023-05-26 11:12:27': 1
},
'response_time': {
'2023-05-26 11:12:27': 0.1
}
},
'model_id': 'test_model',
'prediction_confidence': 0.9,
'prediction_store_policy': 'lts:1'
@@ -404,9 +410,18 @@ async def test_format_prediction_with_timestamp_erl(gates_activity):
# Arrange
input_data = {
**metadata,
'data': {'prediction': [1, 2, 3],
'response_time': [0.1, 0.2, 0.3],
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
'data': {
'prediction': {
'2023-05-26 11:12:27': 1,
'2023-05-26 11:12:28': 2,
'2023-05-26 11:12:29': 3,
},
'response_time': {
'2023-05-26 11:12:27': 0.1,
'2023-05-26 11:12:28': 0.2,
'2023-05-26 11:12:29': 0.3,
}
},
'model_id': 'test_model',
'prediction_confidence': 0.9,
'prediction_store_policy': 'erl:2'
@@ -431,9 +446,18 @@ async def test_format_prediction_with_timestamp_lts(gates_activity):
# Arrange
input_data = {
**metadata,
'data': {'prediction': [1, 2, 3],
'response_time': [0.1, 0.2, 0.3],
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29']},
'data': {
'prediction': {
'2023-05-26 11:12:27': 1,
'2023-05-26 11:12:28': 2,
'2023-05-26 11:12:29': 3,
},
'response_time': {
'2023-05-26 11:12:27': 0.1,
'2023-05-26 11:12:28': 0.2,
'2023-05-26 11:12:29': 0.3,
}
},
'model_id': 'test_model',
'prediction_confidence': 0.9,
'prediction_store_policy': 'lts:2'

20
tests/laborious/utils/repository/test_model_repository.py
View File

@@ -59,7 +59,12 @@ def test_transform_error(mlflow_repository):
def test_predict_success(mlflow_repository):
data = 'data'
data = DataFrame({
'feat_1': {
'index_1': 2,
'index_2': 3
}
})
model_name = 'model'
mlflow_repository.model_serving.get_cached_predict.return_value = np.array(
[2, 3]
@@ -71,10 +76,15 @@ def test_predict_success(mlflow_repository):
model_name, data, 1)
assert output['success'] is True
assert output['content'] == {'prediction': {
0: 2,
1: 3
}, 'response_time': ANY}
assert output['content'] == {
'prediction': {
'index_1': 2,
'index_2': 3
}, 'response_time': {
'index_1': ANY,
'index_2': ANY
}
}
def test_predict_error(mlflow_repository):

4
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images.
pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion.
tag: "0.4.4"
tag: "0.4.5"
# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
imagePullSecrets:
@@ -151,7 +151,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1205-alterar-opc-para-assincrono"
value: "SIENTIAPDE-1182-ajustar-laborious-para-pegar-timestamp-da-resposta-do-mlflow"
- name: PYTHON_APP
value: "laborious.worker.worker"