from collections import namedtuple
from enum import IntFlag
from typing import Any
from collections.abc import Callable
import pandas
from pandas import DataFrame
from ppc_robot_lib.output import charts, iter
from ppc_robot_lib.tasks import task_context
from ppc_robot_lib.utils.types import CellStyle, Column, ColumnGroup
MergeDef = namedtuple('MergeDef', ['start_row', 'end_row', 'start_column', 'end_column'])
"""
Definition of cell merge. All values are indexes (starting at zero) of rows or columns as follows:
* ``start_row`` is inclusive index of row at which the merge starts.
* ``end_row`` is exclusive index of row where the merge ends. This means that for a merge that happens in a single row,
value of ``start_row`` + 1 will be used.
* ``start_column`` is inclusive index of column, similar to the ``start_row`` field.
* ``end_column`` is exclusive index of column, similar to the ``end_row`` field.
"""
ListBuilderFunc = Callable[[DataFrame, 'task_context.TaskContextInterface'], list]
DataBuilderFunc = Callable[[list, DataFrame, list[Column], 'task_context.TaskContextInterface'], None]
class AbsoluteReference(IntFlag):
ROW = 1
COLUMN = 2
[docs]
class Table:
"""
Definition of a table that will be written to output. It defines internal name of the table, output sheet,
table header, a function used to transform the data, number of columns to freeze and whether
to freeze top rows containing header or not.
The header will be used as colum mapping and to generate the header. It is a collection of
:py:class:`ppc_robot_lib.utils.types.Column` or :py:class:`ppc_robot_lib.utils.types.ColumnGroup` instances.
``ColumnGroup`` can contain multiple columns and even another ``ColumnGroup``. These groups will be used
to generate multi-row headers.
Each column must have at least title and number of column to use. You can also set style for the header
and format for the values. See the :py:class:`ppc_robot_lib.utils.types.Column` and
:py:class:`ppc_robot_lib.utils.types.ColumnGroup` classes for more details. The output adapter might
ignore styles, or tailor them to its needs. They are best suited for Google Spreadsheet and Excel outputs.
If no header is given, it shall be derived from input table.
**Header example:**
>>> from ppc_robot_lib.utils.types import Color, Column, ColumnGroup, Format, FormatType
>>> percent_fmt = Format(FormatType.PERCENT, pattern="#.00%")
>>> money_fmt = Format(FormatType.CURRENCY)
>>> header = [
... Column('Account Name', 'AccountDescriptiveName'),
... Column('Account Number', 'AccountNumber'),
... ColumnGroup('PNO', columns=[
... Column('PNO Yday', 'Pno_yesterday', cell_format=percent_fmt),
... Column('PNO 7 Days', 'Pno_last_7_days', cell_format=percent_fmt),
... Column('PNO 30 Days', 'Pno_last_30_days', cell_format=percent_fmt),
... ]),
... ColumnGroup('Conversion Value', columns=[
... Column('Value Yesterday', 'AllConversionValue_yesterday', cell_format=money_fmt),
... Column('Value Last Week', 'AllConversionValue_last_week', cell_format=money_fmt),
... Column('Value This Month', 'AllConversionValue_this_month', cell_format=money_fmt),
... Column('Value Last Month', 'AllConversionValue_last_month', cell_format=money_fmt),
... ]),
... ]
The ``data_builder`` function will recieve the following positional arguments, in order:
#. ``data`` -- empty list of rows. It is the functions's responsibility to fill-in the list. Each row must be
a list of cell values.
#. ``table`` -- :py:class:`pandas.DataFrame` with data.
#. ``row_columns`` -- flattend list of :py:class:`ppc_robot_lib.utils.types.Column`. Can be used to map table
columns to values in rows.
#. ``task_ctx`` -- :py:class:`ppc_robot_lib.tasks.task_context.TaskContextInterface` instance.
Both number and order of columns in output must match the header given in the ``header`` argument.
**Data Builder example:**
>>> def data_builder(data, table, columns, task_ctx):
... for row in table.itertuples():
... data.append([row.CampaignName, row.Impressions])
"""
def __init__(
self,
table_name: str | pandas.DataFrame,
sheet_name: str | Callable[['task_context.TaskContextInterface'], str],
header: list[ColumnGroup | Column] = None,
data_builder: ListBuilderFunc = None,
freeze_header=False,
freeze_columns=0,
sheet_id: int = None,
row_height: int | None = None,
create_filter: bool = False,
currency_column: str = None,
embedded_charts: list['charts.EmbeddedChart'] = None,
):
"""
:param table_name: Name of the table (:py:class:`pandas.DataFrame`) in the current
:py:class:`ppc_robot_lib.work_set.WorkSet` or `pandas.DataFrame` instance.
:param sheet_name: Name of sheet in the output. It can be either a string or a callable that returns string.
:param header: A collection of columns, see example above. If not given, it will be derived from the table.
:param data_builder: Optional function used for generating the output rows.
:param freeze_header: Set to ``True`` if you would like to freeze all header rows.
:param freeze_columns: Number of columns to freeze (row header).
:param sheet_id: Internal sheet identifier. Used for renaming sheets to prevent dropping original one in case
the user changes name of the spreadsheet. Can be any integer that can be represented in JSON.
:param row_height: Row height. If not set, default of the given output platform will be used.
:param create_filter: Create a filterable table in the output sheet. If not given, no filtering will be applied
and no user-changes will be discarded.
:param currency_column: Name of the column that will be used to store the currency.
"""
self.table_name = table_name
self.sheet_name = sheet_name
self.header = header
self.data_builder = data_builder
self.freeze_header = freeze_header
self.freeze_columns = freeze_columns
self.sheet_id = sheet_id
self.row_height = row_height
self.create_filter = create_filter
self.currency_column = currency_column
self.embedded_charts = embedded_charts
[docs]
class SheetData:
"""
Represents a sheet with data ready to be written to output.
"""
def __init__(
self,
sheet_name: str,
row_columns: list[Column],
merges: set[MergeDef],
formats: list[list[CellStyle]],
data: 'iter.SheetDataAdapter',
header_row_count: int,
sheet_id: int,
freeze_header: bool,
freeze_columns: int,
row_height: int | None = None,
create_filter: bool = False,
currency_column_index: int = None,
embedded_charts: list['charts.EmbeddedChart'] = None,
):
"""
:param sheet_name: Sheet name.
:param row_columns: Mapping of rows to columns.
:param merges: Definitions of cell merges in header.
:param formats: Cell styles for the header.
:param data: Complete data matrix, beginning with table header.
:param header_row_count: Number of rows of the header.
:param sheet_id: Sheet identifier.
:param freeze_header: Should the adapter freeze the header rows?
:param freeze_columns: Number of columns to be freezed (will be used as row header).
:param row_height: Row height. If not set, default of the given output platform will be used.
:param create_filter: Create a filterable table in the output sheet. If not given, no filtering will be applied
and no user-changes will be discarded.
:param currency_column_index: Index of a column that will contain currency code for individual rows.
"""
self.sheet_name: str = sheet_name
self.row_columns: list[Column] = row_columns
self.merges: set[MergeDef] = merges
self.formats = formats
self.data = data
self.header_row_count = header_row_count
self.sheet_id = sheet_id
self.freeze_header = freeze_header
self.freeze_columns = freeze_columns
self.row_height = row_height
self.create_filter = create_filter
self.currency_column_index = currency_column_index
self.embedded_charts = embedded_charts
[docs]
class OutputContext:
"""
Context that will be passed to output adapters. Contains current state of the output and state stored
from previous runs (including original ``output_path`` for update).
"""
def __init__(self, state: dict[str, Any], output_path: str | None, previous_state: dict[str, Any] = None):
"""
:param state: Empty dicitionary with state. The adapter can use this dictionary to store state between runs.
The dictionary is always empty and will be stored. On the next run, the dictionary will be passed in
the ``previous_state`` parameter. Everything you store here must be JSON-serializable.
:param output_path: Output path. Can be updated by the adapter, will be stored in the database.
:param previous_state: Read-only state from previous run. Will be ``None`` if this is the first run.
"""
self.state = state
self._output_path = output_path
self._previous_state = previous_state
@property
def output_path(self):
"""
:return: Output path that might be used for update.
"""
return self._output_path
@output_path.setter
def output_path(self, value):
"""
:param value: New output path.
"""
self._output_path = value
@property
def previous_state(self):
"""
:return: Read-only state from previous run. Might be ``None``.
"""
return self._previous_state