SIENTIAPDE-1811

SIENTIAPDE-1811 Implement OPC write error handling and refactor tag writing logic

- Introduced a new function `_apply_opc_write_error` to manage session and reconnect flags based on OPC write error responses.
- Refactored the `_write_tags_from_config` method to streamline the writing of OPC tags for both prediction and confidence data.
- Enhanced unit tests to cover various scenarios for OPC write errors, including session bad and reconnect in progress cases.
- Updated existing tests to validate the new logic and ensure robust error handling.

laborious/activities/opc.py

@@ -27,6 +27,35 @@ def _opc_session_bad_comment(opc_status: str | None) -> str:
     return f'{OPC_SESSION_BAD_COMMENT_PREFIX} {status}'
 
 
+def _apply_opc_write_error(
+    error_info: dict[str, Any] | None,
+    session_bad_seen: bool,
+    session_bad_status: str | None,
+    reconnect_in_progress_seen: bool,
+) -> tuple[bool, str | None, bool]:
+    """
+    Update session/reconnect flags from an OPC write error payload.
+
+    Args:
+        error_info: Repository error details, or None when the write succeeded.
+        session_bad_seen: Whether a session_bad error was seen so far.
+        session_bad_status: Last known OPC status for session errors.
+        reconnect_in_progress_seen: Whether reconnect_in_progress was seen so far.
+
+    Return:
+        Updated (session_bad_seen, session_bad_status, reconnect_in_progress_seen).
+    """
+    if not error_info:
+        return session_bad_seen, session_bad_status, reconnect_in_progress_seen
+
+    kind = error_info.get('opc_error_kind')
+    if kind == 'session_bad':
+        return True, error_info.get('opc_status', session_bad_status), reconnect_in_progress_seen
+    if kind == 'reconnect_in_progress':
+        return session_bad_seen, session_bad_status, True
+    return session_bad_seen, session_bad_status, reconnect_in_progress_seen
+
+
 class OPC(SientiaMonitoring):
     """
     OPC server integration activities for real-time data export.
@@ -198,6 +227,62 @@ class OPC(SientiaMonitoring):
             return False
         return True
 
+    async def _write_tags_from_config(
+        self,
+        server_id: str,
+        tags_config: dict[str, dict[str, Any]],
+        data: DataFrame,
+        data_column: str,
+        tag_type: str,
+        log_label: str,
+        metadata: dict[str, Any],
+    ) -> tuple[dict[str, float | None], bool, str | None, bool]:
+        """
+        Write a group of OPC tags and collect response times and error flags.
+
+        Args:
+            server_id: Target OPC server identifier.
+            tags_config: Tag name to configuration mapping.
+            data: DataFrame with prediction/confidence columns.
+            data_column: Column name whose first row value is written.
+            tag_type: Tag category passed to write_data ('prediction' or 'confidence').
+            log_label: Human-readable label for success logs.
+            metadata: Context metadata for logging and notifications.
+
+        Return:
+            (response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen)
+        """
+        response_times: dict[str, float | None] = {}
+        session_bad_seen = False
+        session_bad_status: str | None = None
+        reconnect_in_progress_seen = False
+
+        for tag, tag_config in tags_config.items():
+            response_time, error_info = await self.write_data(
+                server_id=server_id,
+                tag=tag,
+                data=data.head(1)[data_column].values[0],
+                data_type=tag_config['data_type'],
+                tag_type=tag_type,
+                metadata=metadata,
+            )
+            session_bad_seen, session_bad_status, reconnect_in_progress_seen = (
+                _apply_opc_write_error(
+                    error_info,
+                    session_bad_seen,
+                    session_bad_status,
+                    reconnect_in_progress_seen,
+                )
+            )
+            if response_time is not None:
+                self.info(
+                    f'{log_label} written to OPC server {server_id} for tag {tag}.',
+                    metadata,
+                )
+            response_times[tag] = response_time
+
+        return response_times, session_bad_seen, session_bad_status, reconnect_in_progress_seen
+
     async def manage_output_tags(
         self,
         server_id: str,
@@ -231,62 +316,40 @@ class OPC(SientiaMonitoring):
                 - overall_success: True if all configured tags were written successfully
                 - total_tags_written: Count of successfully written tags
         """
-
         response_times: dict[str, float | None] = {}
         session_bad_seen = False
         session_bad_status: str | None = None
         reconnect_in_progress_seen = False
 
-        if 'prediction_tags' in config:
+        tag_groups = (
-            for tag, tag_config in config['prediction_tags'].items():
+            ('prediction_tags', 'prediction', 'prediction', 'Prediction data'),
-                response_time, error_info = await self.write_data(
+            ('confidence_tags', 'prediction_confidence', 'confidence', 'Confidence data'),
-                    server_id=server_id,
+        )
-                    tag=tag,
+        for config_key, data_column, tag_type, log_label in tag_groups:
-                    data=data.head(1)['prediction'].values[0],
+            if config_key not in config:
-                    data_type=tag_config['data_type'],
+                continue
-                    tag_type='prediction',
+            (
-                    metadata=metadata,
+                group_times,
-                )
+                group_session_bad,
-                if error_info:
+                group_status,
-                    kind = error_info.get('opc_error_kind')
+                group_reconnect,
-                    if kind == 'session_bad':
+            ) = await self._write_tags_from_config(
-                        session_bad_seen = True
+                server_id=server_id,
-                        session_bad_status = error_info.get('opc_status', session_bad_status)
+                tags_config=config[config_key],
-                    elif kind == 'reconnect_in_progress':
+                data=data,
-                        reconnect_in_progress_seen = True
+                data_column=data_column,
-                if response_time is not None:
+                tag_type=tag_type,
-                    self.info(
+                log_label=log_label,
-                        f'Prediction data written to OPC server {server_id} for tag {tag}.',
+                metadata=metadata,
-                        metadata,
+            )
-                    )
+            response_times.update(group_times)
-                response_times[tag] = response_time
+            if group_session_bad:
-
+                session_bad_seen = True
-        if 'confidence_tags' in config:
+                session_bad_status = group_status or session_bad_status
-            for tag, tag_config in config['confidence_tags'].items():
+            if group_reconnect:
-                response_time, error_info = await self.write_data(
+                reconnect_in_progress_seen = True
-                    server_id=server_id,
-                    tag=tag,
-                    data=data.head(1)['prediction_confidence'].values[0],
-                    data_type=tag_config['data_type'],
-                    tag_type='confidence',
-                    metadata=metadata,
-                )
-                if error_info:
-                    kind = error_info.get('opc_error_kind')
-                    if kind == 'session_bad':
-                        session_bad_seen = True
-                        session_bad_status = error_info.get('opc_status', session_bad_status)
-                    elif kind == 'reconnect_in_progress':
-                        reconnect_in_progress_seen = True
-                if response_time is not None:
-                    self.info(
-                        f'Confidence data written to OPC server {server_id} for tag {tag}.',
-                        metadata,
-                    )
-                response_times[tag] = response_time
 
         success = None not in response_times.values()
-
         return (
             success,
             response_times,

tests/laborious/activities/test_opc.py

@@ -13,6 +13,7 @@ from laborious.activities.opc import (
     OPC_SESSION_BAD_CONFIDENCE,
     OPC_WRITTING_ERROR_CONFIDENCE,
     OPC_WRITTING_ERROR_MESSAGE,
+    _apply_opc_write_error,
 )
 
 metadata = {
@@ -289,17 +290,211 @@ async def test_write_data_exception(opc):
         raise AssertionError('Expected an exception to be raised')
 
 
+@mark.parametrize(
+    'error_info,initial_seen,initial_status,initial_reconnect,expected',
+    [
+        (None, False, None, False, (False, None, False)),
+        ({}, False, None, False, (False, None, False)),
+        (
+            {'opc_error_kind': 'session_bad', 'opc_status': 'BadSessionIdInvalid'},
+            False,
+            None,
+            False,
+            (True, 'BadSessionIdInvalid', False),
+        ),
+        (
+            {'opc_error_kind': 'session_bad', 'opc_status': 'NewStatus'},
+            True,
+            'OldStatus',
+            False,
+            (True, 'NewStatus', False),
+        ),
+        (
+            {'opc_error_kind': 'session_bad'},
+            True,
+            'KeptStatus',
+            False,
+            (True, 'KeptStatus', False),
+        ),
+        (
+            {'opc_error_kind': 'reconnect_in_progress'},
+            False,
+            None,
+            False,
+            (False, None, True),
+        ),
+        (
+            {'opc_error_kind': 'other'},
+            True,
+            'Status',
+            True,
+            (True, 'Status', True),
+        ),
+    ],
+)
+def test_apply_opc_write_error(
+    error_info, initial_seen, initial_status, initial_reconnect, expected
+):
+    result = _apply_opc_write_error(
+        error_info,
+        initial_seen,
+        initial_status,
+        initial_reconnect,
+    )
+    assert result == expected
+
+
+@mark.asyncio
+async def test_write_tags_from_config_prediction_success(opc):
+    opc.write_data = AsyncMock(return_value=(0.1, None))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+    tags_config = {'tag1': {'data_type': 'float'}}
+
+    response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
+        server_id='server1',
+        tags_config=tags_config,
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': 0.1}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+    opc.write_data.assert_called_once_with(
+        server_id='server1',
+        tag='tag1',
+        data=0.75,
+        data_type='float',
+        tag_type='prediction',
+        metadata=metadata['metadata'],
+    )
+
+
+@mark.asyncio
+async def test_write_tags_from_config_confidence_success(opc):
+    opc.write_data = AsyncMock(return_value=(0.2, None))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+    tags_config = {'tag2': {'data_type': 'float'}}
+
+    response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
+        server_id='server1',
+        tags_config=tags_config,
+        data=data,
+        data_column='prediction_confidence',
+        tag_type='confidence',
+        log_label='Confidence data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag2': 0.2}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+    opc.write_data.assert_called_once_with(
+        server_id='server1',
+        tag='tag2',
+        data=0.95,
+        data_type='float',
+        tag_type='confidence',
+        metadata=metadata['metadata'],
+    )
+
+
+@mark.asyncio
+async def test_write_tags_from_config_write_failure(opc):
+    opc.write_data = AsyncMock(return_value=(None, {}))
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is False
+
+
+@mark.asyncio
+async def test_write_tags_from_config_session_bad(opc):
+    opc.write_data = AsyncMock(
+        return_value=(
+            None,
+            {
+                'opc_error_kind': 'session_bad',
+                'opc_status': 'BadSessionIdInvalid',
+            },
+        )
+    )
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is True
+    assert opc_status == 'BadSessionIdInvalid'
+    assert reconnect is False
+
+
+@mark.asyncio
+async def test_write_tags_from_config_reconnect_in_progress(opc):
+    opc.write_data = AsyncMock(
+        return_value=(
+            None,
+            {'opc_error_kind': 'reconnect_in_progress'},
+        )
+    )
+    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
+
+    response_times, session_bad, opc_status, reconnect = await opc._write_tags_from_config(
+        server_id='server1',
+        tags_config={'tag1': {'data_type': 'float'}},
+        data=data,
+        data_column='prediction',
+        tag_type='prediction',
+        log_label='Prediction data',
+        metadata=metadata['metadata'],
+    )
+
+    assert response_times == {'tag1': None}
+    assert session_bad is False
+    assert opc_status is None
+    assert reconnect is True
+
+
 @mark.asyncio
 async def test_manage_output_tags_success(opc):
-    opc.write_data = AsyncMock(return_value=(0.1, None))
+    opc._write_tags_from_config = AsyncMock(
-
+        side_effect=[
+            ({'tag1': 0.1}, False, None, False),
+            ({'tag2': 0.1}, False, None, False),
+        ]
+    )
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
     config = {
         'prediction_tags': {'tag1': {'data_type': 'float'}},
         'confidence_tags': {'tag2': {'data_type': 'float'}},
     }
 
-    output_data, opc_metrics, _, _, _ = await opc.manage_output_tags(
+    output_data, opc_metrics, session_bad, opc_status, reconnect = await opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
@@ -308,89 +503,53 @@ async def test_manage_output_tags_success(opc):
 
     assert output_data is True
     assert opc_metrics == {'tag1': 0.1, 'tag2': 0.1}
-    opc.write_data.assert_has_calls(
+    assert session_bad is False
-        [
+    assert opc_status is None
-            call(
+    assert reconnect is False
-                server_id='server1',
+    assert opc._write_tags_from_config.await_count == 2
-                tag='tag1',
-                data=0.75,
-                data_type='float',
-                tag_type='prediction',
-                metadata=metadata['metadata'],
-            ),
-            call(
-                server_id='server1',
-                tag='tag2',
-                data=0.95,
-                data_type='float',
-                tag_type='confidence',
-                metadata=metadata['metadata'],
-            ),
-        ]
-    )
 
 
 @mark.asyncio
-@mark.parametrize(
+async def test_manage_output_tags_failed(opc):
-    'side_effect',
+    opc._write_tags_from_config = AsyncMock(
-    [
+        side_effect=[
-        [(0.1, None), (None, {})],
+            ({'tag1': 0.1}, False, None, False),
-        [(None, {}), (0.2, None)],
+            ({'tag2': None}, False, None, False),
-    ],
+        ]
-)
+    )
-async def test_manage_output_tags_failed(opc, side_effect):
-    opc.write_data = AsyncMock(side_effect=side_effect)
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
     config = {
         'prediction_tags': {'tag1': {'data_type': 'float'}},
         'confidence_tags': {'tag2': {'data_type': 'float'}},
     }
+
     output_data, opc_metrics, _, _, _ = await opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
         metadata=metadata['metadata'],
     )
+
     assert output_data is False
-    assert opc_metrics == {'tag1': side_effect[0][0], 'tag2': side_effect[1][0]}
+    assert opc_metrics == {'tag1': 0.1, 'tag2': None}
-    opc.write_data.assert_has_calls(
-        [
-            call(
-                server_id='server1',
-                tag='tag1',
-                data=0.75,
-                data_type='float',
-                tag_type='prediction',
-                metadata=metadata['metadata'],
-            ),
-            call(
-                server_id='server1',
-                tag='tag2',
-                data=0.95,
-                data_type='float',
-                tag_type='confidence',
-                metadata=metadata['metadata'],
-            ),
-        ]
-    )
 
 
 @mark.asyncio
 async def test_manage_output_tags_do_nothing(opc):
-    opc.write_data = AsyncMock(return_value=(0.1, None))
+    opc._write_tags_from_config = AsyncMock()
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
-    config = {
+    config = {'_invalid_key': {'tag1': {'data_type': 'float'}}}
-        '_invalid_key': {'tag1': {'data_type': 'float'}},
+
-    }
     output_data, opc_metrics, _, _, _ = await opc.manage_output_tags(
         server_id='server1',
         config=config,
         data=data,
         metadata=metadata['metadata'],
     )
+
     assert output_data is True
     assert opc_metrics == {}
-    opc.write_data.assert_not_called()
+    opc._write_tags_from_config.assert_not_called()
 
 
 @mark.asyncio
@@ -507,21 +666,11 @@ def test_process_confidence_generic_failure(opc):
 
 
 @mark.asyncio
-async def test_manage_output_tags_session_bad(opc):
+async def test_manage_output_tags_merges_error_flags(opc):
-    opc.write_data = AsyncMock(
+    opc._write_tags_from_config = AsyncMock(
         side_effect=[
-            (
+            ({'tag1': None}, True, 'BadSessionIdInvalid', False),
-                None,
+            ({'tag2': 0.2}, False, None, True),
-                {
-                    'opc_error_kind': 'session_bad',
-                    'opc_status': 'BadSessionIdInvalid',
-                    'notification_id': 'OPC_WRITE_DATA_ERROR_1',
-                    'message': 'bad',
-                    'block': 'opc_repository',
-                    'level': NotificationLevel.ERROR,
-                },
-            ),
-            (0.2, None),
         ]
     )
     data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
@@ -529,6 +678,7 @@ async def test_manage_output_tags_session_bad(opc):
         'prediction_tags': {'tag1': {'data_type': 'float'}},
         'confidence_tags': {'tag2': {'data_type': 'float'}},
     }
+
     (
         success,
         metrics,
@@ -536,42 +686,12 @@ async def test_manage_output_tags_session_bad(opc):
         opc_status,
         reconnect_in_progress,
     ) = await opc.manage_output_tags('server1', config, data, metadata['metadata'])
+
     assert success is False
     assert session_bad_seen is True
-    assert reconnect_in_progress is False
-    assert opc_status == 'BadSessionIdInvalid'
-    assert metrics['tag1'] is None
-    assert metrics['tag2'] == 0.2
-
-
-@mark.asyncio
-async def test_manage_output_tags_reconnect_in_progress(opc):
-    opc.write_data = AsyncMock(
-        return_value=(
-            None,
-            {
-                'opc_error_kind': 'reconnect_in_progress',
-                'notification_id': 'OPC_WRITE_RECONNECT_IN_PROGRESS_1',
-                'message': 'skipped',
-                'block': 'opc_repository',
-                'level': NotificationLevel.WARNING,
-            },
-        )
-    )
-    data = DataFrame({'prediction': [0.75], 'prediction_confidence': [0.95]})
-    config = {'prediction_tags': {'tag1': {'data_type': 'float'}}}
-    (
-        success,
-        metrics,
-        session_bad_seen,
-        opc_status,
-        reconnect_in_progress,
-    ) = await opc.manage_output_tags('server1', config, data, metadata['metadata'])
-    assert success is False
-    assert session_bad_seen is False
     assert reconnect_in_progress is True
-    assert opc_status is None
+    assert opc_status == 'BadSessionIdInvalid'
-    assert metrics['tag1'] is None
+    assert metrics == {'tag1': None, 'tag2': 0.2}
 
 
 def test_process_confidence_reconnect_in_progress(opc):