from __future__ import annotations
from collections.abc import Callable
import pandas
from ppc_robot_lib import tasks
from ppc_robot_lib.models import Notification, NotificationLevel
from ppc_robot_lib.notifications.exception import CodeType
from ppc_robot_lib.notifications.notification import ParamsType
from ppc_robot_lib.notifications.objects import Table
from ppc_robot_lib.steps import AbstractStep
NumberType = Callable[['tasks.TaskContextInterface'], int] | int
TextParamsType = ParamsType | Callable[['tasks.TaskContextInterface'], ParamsType]
[docs]
class NotificationOutputStep(AbstractStep):
"""
Issues a notification of the given conditions are met.
Several prepared conditions from the :py:mod:`ppc_robot_lib.steps.conditions` module can be used.
Example::
>>> from ppc_robot_lib.steps import conditions
>>> from ppc_robot_lib.steps.output import NotificationOutputStep
>>> from ppc_robot_lib.steps.transformations import GroupByAndAggregateStep, RenameStep
>>> # Table errors contains two coumns: ``url`` and ``error`` text.
... # The following code groups it by the error text.
>>> yield GroupByAndAggregateStep('errors', 'error', {
... 'url': 'count',
... }, 'notifications_error')
... yield RenameStep('notifications_error', {'error': 'Error', 'url': 'Number of URLs'})
>>> def get_error_count(task_ctx):
... return task_ctx.work_set.get_table('notifications')['Number of URLs'].sum()
>>> def get_text_params(task_ctx):
... return [get_error_count(task_ctx)]
>>> yield NotificationOutputStep(
... condition=conditions.table_not_empty('notifications'),
... text='URL checker found %d error.',
... text_plural='URL Checker found %d errors.',
... text_number=get_error_count,
... text_params=get_text_params,
... category=4,
... code=0,
... details_table='notifications'
... )
"""
def __init__(
self,
condition: Callable[[tasks.TaskContextInterface], bool],
text,
code: CodeType,
category=None,
text_localized=True,
text_plural=None,
text_number: NumberType = None,
text_params: TextParamsType = None,
score: NumberType = None,
level=NotificationLevel.WARNING,
details_table=None,
entity_id_column=None,
):
"""
:param condition: Condition -- callable that receives a ``TaskContext`` and returns ``bool``. If ``True``
is returned, the notification issued.
:param text: Notification text.
:param code: Notification code -- can be either int, or reference to
:py:class:`ppc_robot_lib.exceptions.FactoryMethod`. If a ``FactoryMethod`` is given, code and category
is extracted from its definition. See :doc:`/notifications/categories/index` for list of categories
and codes.
:param category: Category code, when not given it is extracted from the object given in the ``code`` argument.
See :doc:`/notifications/categories/index` for list of categories and codes.
:param text_localized: Should the given text be localized in the interface?
:param text_plural: Plural version of the text, used for localization.
:param text_number: Number that will be used to determine which plural form shall be used. Can be given as int,
or callable that receives a ``TaskContext`` and returns int.
:param text_params: Text parameters, if given it will be used as expansion parameters for the ``%`` operator
when rendering the text. It can be a ``dict``, ```list``/```tuple``, or a callable that receives
``TaskContext`` and returns either of the types.
:param score: Overall score used to rank notifications. Higher score means more important (severe) notification.
The score can be given either as an int, or a callable that receives a ``TaskContext`` and returns int.
:param level: One of :py:class:`ppc_robot_lib.models.notification.NotificationLevel` constants.
:param details_table: Name of the table that will be attached to the notification.
:param entity_id_column: IDs of entities that are related to the given notification.
"""
self.condition = condition
if category is None:
if hasattr(code, 'code') and hasattr(code, 'category'):
self.code = code.code
self.category = code.category
else:
raise ValueError(
'You have to specify notification category, or provide an object with both code and category '
'attributes in the code parameter.'
)
else:
self.category = category
self.code = code
self.text = text
self.text_localized = text_localized
self.text_plural = text_plural
self.text_number = text_number
self.text_params = text_params
self.score = score
self.level = level
self.details_table = details_table
self.entity_id_column = entity_id_column
def execute(self, task_ctx: tasks.TaskContextInterface) -> tasks.StepPerformance | None:
condition_result = self.condition(task_ctx)
if condition_result:
notification = self.create_notification(task_ctx)
task_ctx.add_notification(notification)
return None
def create_notification(self, task_ctx: tasks.TaskContextInterface) -> Notification:
if callable(self.score):
score = self.score(task_ctx)
elif self.score:
score = self.score
else:
score = 1
if callable(self.text_number):
text_number = self.text_number(task_ctx)
else:
text_number = self.text_number
if callable(self.text_params):
text_params = self.text_params(task_ctx)
else:
text_params = self.text_params
if self.details_table:
table = task_ctx.work_set.get_table(self.details_table)
details = [self.create_details_table(table, self.entity_id_column)]
else:
details = None
notification = Notification(
level=self.level,
category=self.category,
code=self.code,
score=int(score),
localized=self.text_localized,
text=self.text,
text_plural=self.text_plural,
text_number=int(text_number) if text_number else None,
)
notification.text_parameters = text_params
notification.details_objects = details
return notification
@staticmethod
def create_details_table(table: pandas.DataFrame, id_column=None):
return Table.from_dataframe(table, id_column)