from typing import Any
import pandas
from ppc_robot_lib import tasks
from ppc_robot_lib.models import Notification
from ppc_robot_lib.notifications import factory
from ppc_robot_lib.reporting.transformation import filter_table
class EmitNotification(factory.NotificationFactory):
"""
In current task context emit a :class:`Notification`.
**Note:** Use aliases: ``emit_error_notification``, ``emit_warning_notification``
and ``emit_info_notification``, which explicitly defines notification level.
"""
def __call__(self) -> Notification:
task_ctx = tasks.get_context()
notification = super().__call__()
task_ctx.add_notification(notification)
return notification
class EmitErrorNotification(EmitNotification, factory.NotificationErrorFactory):
pass
class EmitWarningNotification(EmitNotification, factory.NotificationWarningFactory):
pass
class EmitInfoNotification(EmitNotification, factory.NotificationInfoFactory):
pass
emit_notification = EmitNotification
emit_error_notification = EmitErrorNotification
emit_warning_notification = EmitWarningNotification
emit_info_notification = EmitInfoNotification
[docs]
def emit_custom_notifications(
table: pandas.DataFrame, notifications: list[dict[str, Any]], filter_factory
) -> list[Notification]:
"""
Emits notifications specified by the user via the UI.
Each notification has a condition that must be met in order to emit the notification.
:param table: Data Frame that will be queries in the conditions.
:param notifications: List of user-defined notifications.
:param filter_factory: Filter factory that will be used to perform in-memory filtering when checking whether
the conditions are met.
:return:
"""
results = []
for notification in notifications:
if _should_emit_notification(table, notification, filter_factory):
entity = emit_warning_notification(
notification['text'],
code=0,
category=notification.get('category', 11),
text_localized=False,
score=1,
)
results.append(entity)
return results
def _should_emit_notification(table: pandas.DataFrame, notification: dict[str, Any], filter_factory) -> bool:
"""
Checks the conditions for the given notification.
:param table: Table to query in the condition.
:param notification: Notification definition.
:param filter_factory: Filter factory that will be used to perform in-memory filtering when checking whether
the conditions are met.
:return: Are the conditions required by the notification met?
"""
cond_type = notification['type']
if cond_type == 'table_not_empty':
return len(table) > 0
elif cond_type == 'table_empty':
return len(table) == 0
elif cond_type == 'rows_match':
count = notification.get('count', 1)
conditions = notification.get('conditions', [])
if conditions:
filters = filter_factory(conditions)
filtered_table = filter_table(table, filters)
else:
filtered_table = table
return len(filtered_table) >= count
else:
raise ValueError(f'Unknown notification condition: {repr(cond_type)}')