diff --git a/laborious/activities/opc.py b/laborious/activities/opc.py index 524237e..be5a941 100644 --- a/laborious/activities/opc.py +++ b/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: - for tag, tag_config in config['prediction_tags'].items(): - response_time, error_info = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - 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'Prediction data written to OPC server {server_id} for tag {tag}.', - metadata, - ) - response_times[tag] = response_time - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - response_time, error_info = await self.write_data( - 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 + tag_groups = ( + ('prediction_tags', 'prediction', 'prediction', 'Prediction data'), + ('confidence_tags', 'prediction_confidence', 'confidence', 'Confidence data'), + ) + for config_key, data_column, tag_type, log_label in tag_groups: + if config_key not in config: + continue + ( + group_times, + group_session_bad, + group_status, + group_reconnect, + ) = await self._write_tags_from_config( + server_id=server_id, + tags_config=config[config_key], + data=data, + data_column=data_column, + tag_type=tag_type, + log_label=log_label, + metadata=metadata, + ) + response_times.update(group_times) + if group_session_bad: + session_bad_seen = True + session_bad_status = group_status or session_bad_status + if group_reconnect: + reconnect_in_progress_seen = True success = None not in response_times.values() - return ( success, response_times, diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py index eaf946f..c2d37dd 100644 --- a/tests/laborious/activities/test_opc.py +++ b/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( - [ - 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'], - ), - ] - ) + assert session_bad is False + assert opc_status is None + assert reconnect is False + assert opc._write_tags_from_config.await_count == 2 @mark.asyncio -@mark.parametrize( - 'side_effect', - [ - [(0.1, None), (None, {})], - [(None, {}), (0.2, None)], - ], -) -async def test_manage_output_tags_failed(opc, side_effect): - opc.write_data = AsyncMock(side_effect=side_effect) +async def test_manage_output_tags_failed(opc): + opc._write_tags_from_config = AsyncMock( + side_effect=[ + ({'tag1': 0.1}, False, None, False), + ({'tag2': None}, 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( 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]} - 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'], - ), - ] - ) + assert opc_metrics == {'tag1': 0.1, 'tag2': None} @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 = { - '_invalid_key': {'tag1': {'data_type': 'float'}}, - } + config = {'_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): - opc.write_data = AsyncMock( +async def test_manage_output_tags_merges_error_flags(opc): + opc._write_tags_from_config = AsyncMock( side_effect=[ - ( - None, - { - '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), + ({'tag1': None}, True, 'BadSessionIdInvalid', False), + ({'tag2': 0.2}, False, None, True), ] ) 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 metrics['tag1'] is None + assert opc_status == 'BadSessionIdInvalid' + assert metrics == {'tag1': None, 'tag2': 0.2} def test_process_confidence_reconnect_in_progress(opc):