SIENTIAPDE-2072: bump sientia_do pin to 1.12.2

Picks up the notification timestamp -> native datetime fix.

e2e/test_drift.py

@@ -142,7 +142,7 @@ def _recent_minute_timestamps(count: int, offset_minutes: int = 6) -> list[str]:
 def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None:
     """
     Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns
-    ``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``.
+    ``reference_rows`` by writing them to ``dst_path/retrain_input.csv``.
 
     Args:
      - mlflow_repository_stub: External MLflow repository fixture.
@@ -150,12 +150,16 @@ def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFram
     """
 
     def _download(run_id: str, artifact_path: str, dst_path: str, metadata=None):
-        target = Path(dst_path) / 'evaluation_data.csv'
+        target = Path(dst_path) / artifact_path
+        target.parent.mkdir(parents=True, exist_ok=True)
         reference_rows.to_csv(target, index=False)
 
     mlflow_repository_stub._client.get_model_version_by_alias.return_value = MagicMock(
         run_id='fake-reference-run'
     )
+    file_info = MagicMock()
+    file_info.path = 'retrain_input.csv'
+    mlflow_repository_stub._client.list_artifacts.return_value = [file_info]
     mlflow_repository_stub.download_artifacts.side_effect = _download
 
 

laborious/activities/mlflow.py

@@ -53,6 +53,7 @@ class MLFlow(SientiaMonitoring):
 
     _MAX_DEBUG_DATAFRAME_ROWS = 100
     _DEFAULT_MODEL_ALIAS = 'production'
+    _REFERENCE_ARTIFACT_CANDIDATES = ('evaluation_data.csv', 'test_data.csv')
 
     def __init__(
         self,
@@ -527,7 +528,21 @@ class MLFlow(SientiaMonitoring):
             )
 
             # Keep heavy model fitting outside MLflow run timing.
-            wrapper.retrain(data)
+            prediction_data = wrapper.retrain(data)
+            prediction_data.rename(columns={target: 'prediction'}, inplace=True)
+
+            # Merge prediction data with retrain data
+            evaluation_data = pd.merge(
+                data, prediction_data, left_index=True, right_index=True, how='left'
+            )
+
+            # Rename target column to "target"
+            evaluation_data.rename(columns={target: 'target'}, inplace=True)
+
+            # Reset index and put as column "timestamp"
+            evaluation_data['timestamp'] = evaluation_data.index
+            evaluation_data.reset_index(drop=True, inplace=True)
+            evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True)
 
             run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}'
 
@@ -541,8 +556,12 @@ class MLFlow(SientiaMonitoring):
                 tmp_dir = tempfile.mkdtemp(prefix='laborious_retrain_')
                 try:
                     raw_csv = Path(tmp_dir) / 'retrain_input.csv'
+                    evaluation_csv = Path(tmp_dir) / 'evaluation_data.csv'
                     data.to_csv(raw_csv, index=False)
+                    evaluation_data.to_csv(evaluation_csv, index=False)
+
                     mlflow.log_artifact(str(raw_csv))
+                    mlflow.log_artifact(str(evaluation_csv))
                 finally:
                     rmtree(tmp_dir, ignore_errors=True)
 
@@ -639,21 +658,59 @@ class MLFlow(SientiaMonitoring):
             self.error(trace, metadata=metadata)
             raise e
 
+    def _resolve_reference_artifact_name(self, run_id: str) -> str | None:
+        """
+        Pick the first available reference CSV artifact path from the MLflow run.
+
+        Candidates are checked in priority order: ``retrain_input.csv``, then ``train_data.csv``.
+        A path matches when it equals the candidate or ends with ``/<candidate>`` for nested layouts.
+
+        Args:
+         - run_id: MLflow run UUID linked to the production model version.
+
+        Return:
+            Artifact path string for ``download_artifacts``, or ``None`` if no candidate exists.
+        """
+        listed = self.mlflow_repository._client.list_artifacts(run_id)
+        paths = [file_info.path for file_info in listed]
+        for candidate in self._REFERENCE_ARTIFACT_CANDIDATES:
+            for path in paths:
+                if path == candidate or path.endswith(f'/{candidate}'):
+                    return path
+        return None
+
+    def _find_downloaded_csv(self, tmpdir: str, artifact_name: str) -> Path | None:
+        """
+        Locate a downloaded reference CSV in the temp directory.
+
+        Args:
+         - tmpdir: Directory where ``download_artifacts`` wrote files.
+         - artifact_name: Basename of the resolved artifact (e.g. ``retrain_input.csv``).
+
+        Return:
+            ``Path`` to the CSV file if found, else ``None``.
+        """
+        direct = Path(tmpdir) / artifact_name
+        if direct.exists():
+            return direct
+        matches = list(Path(tmpdir).rglob(artifact_name))
+        return matches[0] if matches else None
+
     @activity.defn(name='get_reference_data')
     def get_reference_data(self, input_data: dict[str, Any]) -> list[dict] | None:
         """
-        Download ``evaluation_data.csv`` from the MLflow run linked to ``production`` and parse it.
+        Download reference training CSV from the MLflow run linked to the production alias.
 
-        Used by drift workflows to compare live data against the reference distribution logged with
+        Resolves ``retrain_input.csv`` or ``train_data.csv`` via artifact listing before download.
-        the model. Artifacts are downloaded to a temp directory, discovered via ``rglob`` (nested
+        ``retrain_input.csv`` is preferred when both exist (most recent retrain snapshot). Used by
-        layout-safe), then timestamps are normalized to ``DATETIME_FORMAT`` string columns before
+        drift workflows to compare live data against the reference distribution logged with the model.
-        returning record-oriented dicts.
+        Timestamps are normalized to ``DATETIME_FORMAT`` string columns before returning records.
 
         Args:
-         - input_data: ``metadata`` and ``model_name`` for registry lookup.
+         - input_data: ``metadata``, ``model_name``, and optional ``model_config`` with ``alias``.
 
         Return:
-            List of row dicts, or ``None`` if the artifact path is missing or any step fails.
+            List of row dicts with normalized timestamps, or ``None`` if resolution or load fails.
         """
         metadata = input_data['metadata']
         model_name = input_data['model_name']
@@ -666,20 +723,25 @@ class MLFlow(SientiaMonitoring):
             )
             run_id = mv.run_id
 
+            artifact_path = self._resolve_reference_artifact_name(run_id)
+            if artifact_path is None:
+                self.warning(f'Reference data not found for model {model_name}', metadata)
+                return None
+
+            artifact_name = Path(artifact_path).name
             tmpdir = tempfile.mkdtemp(prefix='laborious_eval_')
             try:
                 self.mlflow_repository.download_artifacts(
                     run_id=run_id,
-                    artifact_path='evaluation_data.csv',
+                    artifact_path=artifact_path,
                     dst_path=tmpdir,
                     metadata=metadata,
                 )
-                csv_candidates = list(Path(tmpdir).rglob('evaluation_data.csv'))
+                csv_path = self._find_downloaded_csv(tmpdir, artifact_name)
-                if not csv_candidates:
+                if csv_path is None:
                     self.warning(f'Reference data not found for model {model_name}', metadata)
                     return None
-
+                reference_data = pd.read_csv(csv_path)
-                reference_data = pd.read_csv(csv_candidates[0])
             finally:
                 rmtree(tmpdir, ignore_errors=True)
 

laborious/activities/opc.py

@@ -143,9 +143,7 @@ class OPC(SientiaMonitoring):
                     attachment_content=error_data.get('attachment_content', None),
                 )
             else:
-                self.info(
+                self.info(f'OPC server {opc_id}:{server["server_name"]} connected successfully.')
-                    f'OPC server {opc_id}:{server["server_name"]} connected successfully.'
-                )
 
     def write_data(
         self,

laborious/utils/connectors_config.py

@@ -11,8 +11,7 @@ def build_mlflow_config() -> dict[str, Any]:
     same string workers and notebooks should use for ``MLFLOW_TRACKING_URI``-style clients.
 
     Environment Variables:
-        MLFLOW_HOST: Host with scheme; port optional if MLFLOW_PORT is set (default: http://localhost)
+        MLFLOW_URL: Host with scheme
-        MLFLOW_PORT: Appended when host has no explicit port (default: 5080)
         MLFLOW_USERNAME: Basic-auth or service user (default: aignosi)
         MLFLOW_PASSWORD: Password or token (default: aignosi)
 

laborious/utils/repository/opc_repository.py

@@ -755,9 +755,7 @@ class OpcRepository(SientiaMonitoring):
                 pass
         node_obj.set_value(data, variant_type)
 
-    def _write_reconnect_in_progress(
+    def _write_reconnect_in_progress(self, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
-        self, metadata: dict[str, Any]
-    ) -> tuple[bool, dict[str, Any]]:
         """
         Fail a write because a background reconnect thread is already running.
 

laborious/worker/worker.py

@@ -147,9 +147,13 @@ async def main():
         to_install_runtime = runtime
 
     try:
-        await plugin_store.install_runtime(runtime_name=to_install_runtime, metadata=metadata_runtime)
+        await plugin_store.install_runtime(
+            runtime_name=to_install_runtime, metadata=metadata_runtime
+        )
     except Exception as exc:
-        logger.custom_critical(f'Failed to install runtime {to_install_runtime}: {exc}', metadata_runtime)
+        logger.custom_critical(
+            f'Failed to install runtime {to_install_runtime}: {exc}', metadata_runtime
+        )
         metrics.APP_UP.labels(pod_id=POD_ID).set(0)
         sys.exit(1)
 
@@ -212,6 +216,7 @@ async def main():
                 activities.export_data_to_postgres,
             ],
             logger=logger,
+            runtime='core',
         ),
         prepare_worker(
             temporal_client=temporal_client,
@@ -224,6 +229,7 @@ async def main():
                 activities.export_data_to_postgres,
             ],
             logger=logger,
+            runtime='core',
         ),
         prepare_worker(
             temporal_client=temporal_client,

requirements-local.txt

@@ -3,7 +3,7 @@ psycopg2-binary
 sqlalchemy
 asyncua==1.0.6
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
 git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
 prometheus-client   
 botocore

requirements.txt

@@ -3,7 +3,7 @@ psycopg2-binary
 sqlalchemy
 asyncua==1.0.6
 redis
-sientia_do>=1.12.1
+sientia_do>=1.12.2
 sientia_model>=0.8.2
 prometheus-client
 botocore

tests/laborious/activities/test_mlflow.py

@@ -399,6 +399,8 @@ def test_retrain_model_success_data_success_retrain(
             'value': [1.0, 2.0],
         }
     )
+    pivoted_index = pd.DatetimeIndex([ts])
+    wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0])
 
     payload = MagicMock()
     payload.retrieve = MagicMock(return_value=raw_data)
@@ -428,6 +430,7 @@ def test_retrain_model_success_with_payload_data(
     mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
 ):
     mock_mkdtemp.return_value = 'tmp'
+    mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
     wrapper = MagicMock()
@@ -437,20 +440,21 @@ def test_retrain_model_success_with_payload_data(
     mock_cm.__exit__.return_value = False
     mlflow.mlflow_repository.start_run.return_value = mock_cm
 
-    raw_data = MagicMock(columns=['variable', 'timestamp', 'value', 'created_at'])
+    ts = pd.Timestamp('2020-01-01', tz='UTC')
-    raw_data.__getitem__.return_value.max.return_value = 'ts'
+    raw_data = pd.DataFrame(
+        {
+            'variable': ['target', 'f1', 'target', 'f1'],
+            'timestamp': [ts, ts, ts + pd.Timedelta(hours=1), ts + pd.Timedelta(hours=1)],
+            'value': [1.0, 2.0, 3.0, 4.0],
+            'created_at': [ts, ts, ts + pd.Timedelta(hours=1), ts + pd.Timedelta(hours=1)],
+        }
+    )
+    pivoted_index = pd.DatetimeIndex([ts, ts + pd.Timedelta(hours=1)])
+    wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0, 3.0])
+
     payload = MagicMock()
     payload.retrieve = MagicMock(return_value=raw_data)
 
-    pivoted = MagicMock()
-    raw_data.sort_values.return_value = raw_data
-    raw_data.drop_duplicates.return_value = raw_data
-    raw_data.pivot.return_value = pivoted
-    pivoted.fillna = MagicMock()
-    pivoted.columns.name = None
-    pivoted.index = MagicMock()
-    pivoted.__setitem__ = MagicMock()
-
     response = mlflow.retrain_model(
         {
             **metadata,
@@ -482,13 +486,17 @@ def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
     mlflow.mlflow_repository.start_run.return_value = mock_cm
 
     ts = pd.Timestamp('2020-01-01', tz='UTC')
+    ts_next = ts + pd.Timedelta(days=1)
     raw_data = pd.DataFrame(
         {
             'variable': ['target', 'f1', 'target', 'f1'],
-            'timestamp': [ts, ts, ts + pd.Timedelta(days=1), ts + pd.Timedelta(days=1)],
+            'timestamp': [ts, ts, ts_next, ts_next],
             'value': [1.0, 2.0, 3.0, 4.0],
         }
     )
+    pivoted_index = pd.DatetimeIndex([ts, ts_next])
+    wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0, 3.0])
+
     payload = MagicMock()
     payload.retrieve = MagicMock(return_value=raw_data)
 
@@ -652,6 +660,18 @@ def test_update_production_model_error(mlflow):
         raise AssertionError('Expected exception')
 
 
+def _artifact_file_info(path: str) -> MagicMock:
+    file_info = MagicMock()
+    file_info.path = path
+    return file_info
+
+
+def _retrain_prediction_frame(
+    index: pd.DatetimeIndex, target: str, values: list[float]
+) -> pd.DataFrame:
+    return pd.DataFrame({target: values}, index=index)
+
+
 @patch('laborious.activities.mlflow.to_datetime')
 def test_get_reference_data_success(mock_to_datetime, mlflow):
     input_data = {
@@ -661,6 +681,9 @@ def test_get_reference_data_success(mock_to_datetime, mlflow):
 
     mv = MagicMock(run_id='run1')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('evaluation_data.csv'),
+    ]
 
     mock_reference_data = MagicMock()
     mock_reference_data.__getitem__.return_value = MagicMock()
@@ -672,10 +695,19 @@ def test_get_reference_data_success(mock_to_datetime, mlflow):
     with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
         with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
             with patch('laborious.activities.mlflow.rmtree'):
-                with patch('laborious.activities.mlflow.Path') as mp:
+                with patch.object(
-                    mp.return_value.rglob.return_value = [MagicMock()]
+                    mlflow,
+                    '_find_downloaded_csv',
+                    return_value=MagicMock(),
+                ):
                     result = mlflow.get_reference_data(input_data)
 
+    mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
+        run_id='run1',
+        artifact_path='evaluation_data.csv',
+        dst_path='/t',
+        metadata=metadata['metadata'],
+    )
     mock_reference_data.to_dict.assert_called_once_with(orient='records')
     assert result == mock_reference_data.to_dict.return_value
 
@@ -694,6 +726,97 @@ def test_get_reference_data_not_found(mlflow):
     assert result is None
 
 
+def test_get_reference_data_only_test_data_csv(mlflow):
+    input_data = {
+        **metadata,
+        'model_name': 'test_model',
+    }
+    mv = MagicMock(run_id='run1')
+    mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('test_data.csv'),
+    ]
+
+    mock_reference_data = MagicMock()
+    mock_reference_data.__getitem__.return_value = MagicMock()
+    mock_reference_data.to_dict.return_value = [{'timestamp': '2023-05-26 11:12:27', 'value': 1.0}]
+
+    with patch('laborious.activities.mlflow.to_datetime') as mock_to_datetime:
+        mock_to_datetime.return_value.dt.strftime.return_value = MagicMock()
+        with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
+            with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
+                with patch('laborious.activities.mlflow.rmtree'):
+                    with patch.object(
+                        mlflow,
+                        '_find_downloaded_csv',
+                        return_value=MagicMock(),
+                    ):
+                        result = mlflow.get_reference_data(input_data)
+
+    mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
+        run_id='run1',
+        artifact_path='test_data.csv',
+        dst_path='/t',
+        metadata=metadata['metadata'],
+    )
+    assert result == mock_reference_data.to_dict.return_value
+
+
+def test_get_reference_data_prefers_evaluation_data_when_both_listed(mlflow):
+    input_data = {
+        **metadata,
+        'model_name': 'test_model',
+    }
+    mv = MagicMock(run_id='run1')
+    mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('test_data.csv'),
+        _artifact_file_info('evaluation_data.csv'),
+    ]
+
+    mock_reference_data = MagicMock()
+    mock_reference_data.__getitem__.return_value = MagicMock()
+    mock_reference_data.to_dict.return_value = [{'timestamp': '2023-05-26 11:12:27', 'value': 1.0}]
+
+    with patch('laborious.activities.mlflow.to_datetime') as mock_to_datetime:
+        mock_to_datetime.return_value.dt.strftime.return_value = MagicMock()
+        with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
+            with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
+                with patch('laborious.activities.mlflow.rmtree'):
+                    with patch.object(
+                        mlflow,
+                        '_find_downloaded_csv',
+                        return_value=MagicMock(),
+                    ):
+                        result = mlflow.get_reference_data(input_data)
+
+    mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
+        run_id='run1',
+        artifact_path='evaluation_data.csv',
+        dst_path='/t',
+        metadata=metadata['metadata'],
+    )
+    assert result == mock_reference_data.to_dict.return_value
+
+
+def test_get_reference_data_no_candidate_artifacts_returns_none(mlflow):
+    input_data = {
+        **metadata,
+        'model_name': 'test_model',
+    }
+    mv = MagicMock(run_id='run1')
+    mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('other_artifact.csv'),
+    ]
+
+    result = mlflow.get_reference_data(input_data)
+
+    mlflow.mlflow_repository.download_artifacts.assert_not_called()
+    mlflow.warning.assert_called()
+    assert result is None
+
+
 def test_get_reference_data_missing_csv_file_returns_none(mlflow):
     input_data = {
         **metadata,
@@ -701,13 +824,17 @@ def test_get_reference_data_missing_csv_file_returns_none(mlflow):
     }
     mv = MagicMock(run_id='run1')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('evaluation_data.csv'),
+    ]
 
     with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='tmp'):
         with patch('laborious.activities.mlflow.rmtree'):
-            with patch('laborious.activities.mlflow.Path') as mp:
+            with patch.object(mlflow, '_find_downloaded_csv', return_value=None):
-                mp.return_value.rglob.return_value = []
                 result = mlflow.get_reference_data(input_data)
 
+    mlflow.mlflow_repository.download_artifacts.assert_called_once()
+    mlflow.warning.assert_called()
     assert result is None
 
 
@@ -719,6 +846,9 @@ def test_get_reference_data_exception(mlflow):
 
     mv = MagicMock(run_id='run1')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
+    mlflow.mlflow_repository._client.list_artifacts.return_value = [
+        _artifact_file_info('evaluation_data.csv'),
+    ]
     mlflow.mlflow_repository.download_artifacts.side_effect = Exception('dl fail')
 
     result = mlflow.get_reference_data(input_data)

values.yaml

@@ -233,11 +233,15 @@ global:
 runtimes:
   basic:
     replicas: 1
-  single:
+  legacy:
     replicas: 1
     env:
       - name: "GITHUB_BRANCH"
         value: "main"
+      - name: "MLFLOW_HOST"
+        value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local"
+      - name: "MLFLOW_PORT"
+        value: "80"
 
 # -----------------------------------------------------------------------------
 # Chart-level configuration (applies to all runtimes)
@@ -305,8 +309,8 @@ ssh:
 
 # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=<pwd>
 #
-# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.5
+# helm upgrade --install sientia-laborious-worker /home/grezewave/Documents/projects/sientia/sientia-core-applications/sientia-module -n sientia --create-namespace -f ./values.yaml
-#
+
 # kubectl create secret generic git-ssh-key-sientia-laborious-worker \
 #   --namespace sientia \
 #   --from-file=ssh-privatekey=git_key \