Source code for lab.reports

Module that permits generating reports by reading properties files

import collections
from collections import defaultdict
import fnmatch
import logging
import math
import numbers
import os
import sys

import txt2tags

from lab import tools
from lab.reports import markup
from lab.reports.markup import Document, ESCAPE_WORDBREAK

[docs] def arithmetic_mean(values): """Compute the arithmetic mean of a sequence of numbers. >>> arithmetic_mean([20, 30, 70]) 40.0 """ assert None not in values return math.fsum(values) / len(values)
[docs] def geometric_mean(values): """Compute the geometric mean of a sequence of numbers. >>> round(geometric_mean([2, 8]), 2) 4.0 """ assert None not in values exp = 1.0 / len(values) return tools.product([val**exp for val in values])
def finite_sum(values): """Compute the sum of a list of numbers, excluding values of None and 'infinity'. """ return sum(x for x in values if x is not None and x != sys.maxsize) def function_name(f): names = { "arithmetic_mean": "arithmetic mean", "finite_sum": "finite sum", "geometric_mean": "geometric mean", } return names.get(f.__name__, f.__name__) def get_aggregation_function(function, functions): """ Code for backwards compatibility. """ if function and functions: logging.critical( 'You cannot use "function" and "functions" kwargs for ' "Attribute at the same time." ) elif functions: tools.show_deprecation_warning( '"functions" kwarg for Attribute is deprecated. Use ' '"function" instead.' ) if len(functions) > 1: logging.critical("Using multiple aggregation functions is unsupported.") return functions[0] else: return function
[docs] class Attribute(str): """A string subclass for attributes in reports.""" def __new__(cls, name, **kwargs): return str.__new__(cls, name) def __init__( self, name, absolute=False, min_wins=True, function=None, functions=None, scale=None, digits=2, ): """ Use this class if your **custom** attribute needs a non-default value for: * *absolute*: if False, only include tasks for which all task runs have values in a per-domain table (e.g. ``coverage`` is absolute, whereas ``expansions`` is not, because we can't compare algorithms A and B for task X if B has no value for ``expansions``). * *min_wins*: set to True if a smaller value for this attribute is better, to False if a higher value is better and to None if values can't be compared. (E.g., *min_wins* is False for ``coverage``, but it is True for ``expansions``). * *function*: the function used to aggregate values of multiple runs for this attribute, for example, in domain reports. Defaults to :py:func:`sum`. * *functions*: deprecated. Pass a single *function* instead. * *scale*: default scaling. Can be one of "linear", "log" and "symlog". If *scale* is None (default), the reports will choose the scaling. * *digits*: number of digits after the decimal point. The ``downward`` package automatically uses appropriate settings for most attributes. >>> avg_h = Attribute("avg_h", min_wins=False) >>> abstraction_done = Attribute( ... "abstraction_done", absolute=True, min_wins=False ... ) """ self.absolute = absolute self.min_wins = min_wins self.function = ( get_aggregation_function(function, tools.make_list(functions)) or sum ) self.scale = scale self.digits = digits def copy(self, name): return Attribute( name, absolute=self.absolute, min_wins=self.min_wins, function=self.function, scale=self.scale, digits=self.digits, )
[docs] class Report: """ Base class for all reports. """ def __init__(self, attributes=None, format="html", filter=None, **kwargs): """ Inherit from this or a child class to implement a custom report. Depending on the type of output you want to make, you will have to overwrite the :meth:`.write`, :meth:`.get_text` or :meth:`.get_markup` method. *attributes* is the list of attributes you want to include in your report. If omitted, use all numerical attributes. Globbing characters * and ? are allowed. Example: >>> report = Report(attributes=["coverage", "translator_*"]) When a report is made, both the available and the selected attributes are printed on the commandline. *format* can be one of e.g. html, tex, wiki (MediaWiki), doku (DokuWiki), pmw (PmWiki), moin (MoinMoin) and txt (Plain text). Subclasses may allow additional formats. If given, *filter* must be a function or a list of functions that are passed a dictionary of a run's attribute keys and values. Filters must return True, False or a new dictionary. Depending on the returned value, the run is included or excluded from the report, or replaced by the new dictionary, respectively. Filters for properties can be given in shorter form without defining a function. To include only runs where attribute ``foo`` has value v, use ``filter_foo=v``. To include only runs where attribute ``foo`` has value v1, v2 or v3, use ``filter_foo=[v1, v2, v3]``. Filters are applied sequentially, i.e., the first filter is applied to all runs before the second filter is executed. Filters given as ``filter_*`` kwargs are applied *after* all filters passed via the ``filter`` kwarg. Examples: Include only the "cost" attribute in a LaTeX report: >>> report = Report(attributes=["cost"], format="tex") Only include successful runs in the report: >>> report = Report(filter_coverage=1) Only include runs in the report where the initial h value is at most 100: >>> def low_init_h(run): ... return run["initial_h_value"] <= 100 ... >>> report = Report(filter=low_init_h) Only include runs from "blocks" and "barman" with a timeout: >>> report = Report(filter_domain=["blocks", "barman"], filter_search_timeout=1) Add a new attribute: >>> def add_expansions_per_time(run): ... expansions = run.get("expansions") ... time = run.get("search_time") ... if expansions is not None and time: ... run["expansions_per_time"] = expansions / time ... return run ... >>> report = Report( ... attributes=["expansions_per_time"], filter=[add_expansions_per_time] ... ) Rename, filter and sort algorithms: >>> def rename_algorithms(run): ... name = run["algorithm"] ... paper_names = {"lama11": "LAMA 2011", "fdss_sat1": "FDSS 1"} ... run["algorithm"] = paper_names[name] ... return run ... >>> # We want LAMA 2011 to be the leftmost column. >>> # filter_* filters are evaluated last, so we use the updated >>> # algorithm names here. >>> algorithms = ["LAMA 2011", "FDSS 1"] >>> report = Report(filter=rename_algorithms, filter_algorithm=algorithms) """ self.attributes = tools.make_list(attributes) if format not in txt2tags.TARGETS + ["eps", "pdf", "pgf", "png", "py"]: raise ValueError(f"invalid format: {format}") self.output_format = format self.toc = True self.run_filter = tools.RunFilter(filter, **kwargs)
[docs] def __call__(self, eval_dir, outfile): """Make the report. This method is called automatically when the report step is executed. It loads the data and calls :meth:`.write`. *eval_dir* must be a path to an evaluation directory containing a ``properties`` file. The report will be written to *outfile*. """ if not eval_dir.endswith("-eval"): 'The source directory does not end with "-eval". ' "Are you sure this is an evaluation directory?" ) self.eval_dir = os.path.abspath(eval_dir) # It would be nice if we could infer "format" from "outfile", but the # former is needed before the latter is available. # Also we can't add the extension ".format" to "outfile" in case it's # missing, because "outfile" might be a directory. self.outfile = os.path.abspath(outfile) # Map from attribute to type. self._all_attributes = {} self._load_data() self._apply_filter() self._scan_data() # Turn string attributes into instances of Attribute. self.attributes = [self._prepare_attribute(attr) for attr in self.attributes] # Expand glob characters. self.attributes = self._glob_attributes(self.attributes) if not self.attributes:"Available attributes: {', '.join(self.all_attributes)}")"Using all numerical attributes.") self.attributes = self._get_numerical_attributes() self.attributes = sorted(self.attributes) # Check for duplicate attributes to avoid "coverage" overwriting # Attribute("coverage") by accident. counter = collections.Counter(self.attributes) duplicates = [name for name, count in sorted(counter.items()) if count > 1] if duplicates: logging.critical(f"Duplicate attributes detected: {duplicates}") self.write()
def _prepare_attribute(self, attr): if isinstance(attr, Attribute): return attr return Attribute(attr) def _glob_attributes(self, attributes): expanded_attrs = [] for attr in attributes: # Attribute without wildcards. Filtering would reset its options. if attr in self.all_attributes: expanded_attrs.append(attr) continue matches = fnmatch.filter(self.all_attributes, attr) if not matches: logging.warning( f'There is no attribute "{attr}" in the properties file.' ) # Use the attribute options from the pattern for all matches, but # don't try to guess options for attributes that appear in the list. expanded_attrs.extend( [attr.copy(match) for match in matches if match not in attributes] ) if attributes and not expanded_attrs: logging.critical("No attributes match your patterns.") return expanded_attrs @property def all_attributes(self): return sorted(self._all_attributes.keys()) def _get_numerical_attributes(self): return [ attr for attr in self._all_attributes.keys() if self.attribute_is_numeric(attr) ] def attribute_is_numeric(self, attribute): """Return true if the values for *attribute* are ints or floats. If the attribute is None in all runs it may be numeric. """ return self._all_attributes[attribute] is None or issubclass( self._all_attributes[attribute], numbers.Number )
[docs] def get_markup(self): """ Return `txt2tags <>`_ markup for the report. """ table = Table() for run_id, run in self.props.items(): row = {} for key, value in run.items(): if key not in self.attributes: continue if isinstance(value, (list, tuple)): key = "-".join(str(item) for item in value) row[key] = value table.add_row(run_id, row) return str(table)
[docs] def get_text(self): """ Return text (e.g., HTML, LaTeX, etc.) for the report. By default this method calls :meth:`.get_markup` and converts the markup to the desired output *format*. """ name, _ = os.path.splitext(os.path.basename(self.outfile)) doc = Document(title=name) doc.add_text( self.get_markup() or "No tables were generated. " "This happens when no significant changes occured or " "if for all attributes and all problems never all " "algorithms had a value for this attribute in a " "per-domain report." ) return doc.render(self.output_format, {"toc": self.toc})
[docs] def write(self): """ Write the report files. By default this method calls :meth:`.get_text` and writes the obtained text to *outfile*. Overwrite this method if you want to write the report file(s) directly. You should write them to *self.outfile*. """ content = self.get_text() tools.makedirs(os.path.dirname(self.outfile)) tools.write_file(self.outfile, content)"Wrote file://{self.outfile}")
def _get_type(self, attribute): for run in self.props.values(): val = run.get(attribute) if val is not None: return type(val) # Attribute is None in all runs. return None def _get_type_map(self, attributes): return { self._prepare_attribute(attr): self._get_type(attr) for attr in attributes } def _scan_data(self): attributes = set() for run in self.props.values(): attributes |= set(run.keys()) self._all_attributes = self._get_type_map(attributes) def _load_data(self): props_file = os.path.join(self.eval_dir, "properties")"Reading properties file") self.props = tools.Properties(filename=props_file) if not self.props: logging.critical(f"No properties found in {self.eval_dir}")"Reading properties file finished") def _apply_filter(self): self.run_filter.apply(self.props) if not self.props: logging.critical("All runs have been filtered -> Nothing to report.")
class CellFormatter: """Formating information for one cell in a table.""" def __init__( self, bold=False, count=None, link=None, color=None, align_right=False ): self.bold = bold self.count = count = link self.color = color self.align_right = align_right def format_value(self, value): result = str(value) if result = f"[''{result}'' {}]" if self.count: result = f"{result} ({self.count})" if self.bold: result = f"**{result}**" if self.color: result = f"{{{result}|color:{self.color}}}" if self.align_right: result = " " + result else: result += " " return result class Table(collections.defaultdict): def __init__(self, title="", min_wins=None, colored=False, digits=2): """ The *Table* class can be useful for `Report` subclasses that want to return a table as txt2tags markup. It is realized as a dictionary of dictionaries mapping row names to colum names to cell values. To obtain the markup from a table, use the ``str()`` function. *title* will be printed in the top left cell. *min_wins* can be either None, True or False. If it is True (False), the cell with the lowest (highest) value in each row will be highlighted. If *colored* is True, the values of each row will be given colors from a colormap. Numbers are rounded to *digits* positions after the decimal point. >>> t = Table(title="expansions") >>> t.add_cell("prob1", "cfg1", 10) >>> t.add_cell("prob1", "cfg2", 20) >>> t.add_row("prob2", {"cfg1": 15, "cfg2": 25}) >>> def remove_quotes(s): ... return s.replace('""', "") ... >>> print(remove_quotes(str(t))) || expansions | cfg1 | cfg2 | | prob1 | 10 | 20 | | prob2 | 15 | 25 | >>> t.row_names ['prob1', 'prob2'] >>> t.col_names ['cfg1', 'cfg2'] >>> t.get_row("prob2") [15, 25] >>> t.get_columns() == {"cfg1": [10, 15], "cfg2": [20, 25]} True >>> t.add_summary_function("SUM", sum) >>> print(remove_quotes(str(t))) || expansions | cfg1 | cfg2 | | prob1 | 10 | 20 | | prob2 | 15 | 25 | | **SUM** | 25 | 45 | >>> t.set_column_order(["cfg2", "cfg1"]) >>> print(remove_quotes(str(t))) || expansions | cfg2 | cfg1 | | prob1 | 20 | 10 | | prob2 | 25 | 15 | | **SUM** | 45 | 25 | """ collections.defaultdict.__init__(self, dict) self.title = title self.min_wins = min_wins self.row_min_wins = {} self.colored = colored self.digits = digits self.summary_funcs = {} = [] self.num_values = None self.dynamic_data_modules = [] self._cols = None # For printing. self.header_row = "column names (never printed)" self.header_column = "row names (never printed)" self.cell_formatters = collections.defaultdict(dict) self.row_order = None self.column_order = None self.summary_row_order = [] def add_cell(self, row, col, value): """Set Table[row][col] = value.""" self[row][col] = value self._cols = None def add_row(self, row_name, row): """Add a new data row called *row_name* to the table. *row* must be a mapping from column names to values. """ self[row_name] = row self._cols = None def add_col(self, col_name, col): """Add a new data column called *col_name* to the table. *col* must be a mapping from row names to values. """ for row_name, value in col.items(): self[row_name][col_name] = value self._cols = None @property def row_names(self): """Return all data row names in sorted order.""" return self.row_order or tools.natural_sort(self.keys()) @property def col_names(self): """Return all data column names in sorted order.""" if self._cols: return self._cols col_names = set() for row in self.values(): col_names |= set(row.keys()) self._cols = [] if self.column_order: # First use all elements for which we know an order. # All remaining elements will be sorted alphabetically. self._cols = [c for c in self.column_order if c in col_names] col_names -= set(self._cols) self._cols += tools.natural_sort(col_names) return self._cols def get_row(self, row_name): """Return a list of the values in *row*.""" return [self[row_name].get(col_name, None) for col_name in self.col_names] def get_columns(self): """ Return a mapping from column names to the list of values in that column. """ values = defaultdict(list) for row_name in self.row_names: for col_name in self.col_names: values[col_name].append(self[row_name].get(col_name)) return values def add_summary_function(self, name, func): """ Add a bottom row with the values ``func(column_values)`` for each column. *func* can be e.g. :func:`sum`, :func:`arithmetic_mean` or :func:`geometric_mean`. """ self.summary_funcs[name] = func self.summary_row_order.append(name) def set_row_order(self, order): self.row_order = order def set_column_order(self, order): self.column_order = order self._cols = None def get_min_wins(self, row_name=None): """ The table class can store information on whether higher or lower values are better for each row or globally. If no row specific setting for *row_name* is found, the global setting is returned. """ return self.row_min_wins.get(row_name, self.min_wins) def get_summary_rows(self): """ Returns a dictionary mapping names of summary rows to dictionaries mapping column names to values. """ summary_rows = {} for row_name in self.summary_row_order: func = self.summary_funcs[row_name] summary_row = {} for col_name, column in self.get_columns().items(): values = [val for val in column if val is not None] if values: summary_row[col_name] = func(values) else: summary_row[col_name] = None summary_row[self.header_column] = row_name summary_rows[row_name] = summary_row formatter = CellFormatter(bold=True, count=self.num_values) self.cell_formatters[row_name][self.header_column] = formatter return summary_rows def _get_printable_row_order(self): """ Return a list of all rows (including non-data rows) in the order they should be printed. """ row_order = [self.header_row] for row_name in self.row_names + self.summary_row_order: row_order.append(row_name) for module in self.dynamic_data_modules: row_order = module.modify_printable_row_order(self, row_order) or row_order return row_order def _get_printable_column_order(self): """ Return a list of all columns (including non-data columns) in the order they should be printed. """ col_order = [self.header_column] for col_name in self.col_names: col_order.append(col_name) for module in self.dynamic_data_modules: col_order = ( module.modify_printable_column_order(self, col_order) or col_order ) return col_order def _collect_cells(self): """ Collect all cells that should be printed including table headers, row names, summary rows, etc. Returns a dictionary mapping row names to dictionaries mapping column names to values. """ cells = collections.defaultdict(dict) cells[self.header_row][self.header_column] = self.title for col_name in self.col_names: cells[self.header_row][col_name] = str(col_name) # Add data rows and summary rows. for row_name, row in list(self.items()) + list(self.get_summary_rows().items()): cells[row_name][self.header_column] = str(row_name) for col_name in self.col_names: cells[row_name][col_name] = row.get(col_name) for dynamic_data_module in self.dynamic_data_modules: cells = dynamic_data_module.collect(self, cells) or cells return cells def _format(self, cells): """Format all entries in **cells** (in place).""" for row_name, row in cells.items(): self._format_row(row_name, row) for dynamic_data_module in self.dynamic_data_modules: dynamic_data_module.format(self, cells) def _format_value(self, value): if isinstance(value, float): return f"{value:.{self.digits}f}" else: result = str(value) # Only escape text if it doesn't contain LaTeX or HTML markup. if "''" in result: return result else: return markup.escape(result) def _format_row(self, row_name, row): """Format all entries in **row** (in place).""" if row_name == self.header_row: for col_name, value in row.items(): # Allow breaking after underlines. value = value.replace("_", "_" + ESCAPE_WORDBREAK) # Right-align headers (except the left-most one). if col_name != self.header_column: value = " " + value row[col_name] = value return # Get the slice of the row that should be formatted (i.e., the data columns). # Note that there might be other columns (e.g., added by dynamic data # modules) that should not be formatted. row_slice = {col_name: row.get(col_name) for col_name in self.col_names} min_wins = self.get_min_wins(row_name) highlight = min_wins is not None colored = self.colored and highlight if colored: def try_to_round(v): try: return round(v, self.digits) except TypeError: return v rounded_row_slice = { col: try_to_round(val) for col, val in row_slice.items() } colors = tools.get_colors(rounded_row_slice, min_wins) if highlight: min_value, max_value = tools.get_min_max(row_slice.values()) else: min_value, max_value = None, None def is_close(a, b): # Highlight based on precision visible in table, not actual values. return self._format_value(a) == self._format_value(b) for col_name, value in row.items(): color = None bold = False # Format data columns if col_name in row_slice: if colored: color = tools.rgb_fractions_to_html_color(*colors[col_name]) elif ( highlight and value is not None and ( (is_close(value, min_value) and min_wins) or (is_close(value, max_value) and not min_wins) ) ): bold = True row[col_name] = self._format_cell( row_name, col_name, value, color=color, bold=bold ) def _format_cell(self, row_name, col_name, value, color=None, bold=False): """ Return the formatted value for a single cell in the table. *row_name* and *col_name* specify the position of the cell and *value* is the unformatted value. Floats are rounded to two decimal places and lists are quoted. The *color* to render the result in can be given as a string and setting *bold* to true renders the value in bold. If a custom formatter is specified for this cell, it is used instead of this default format. """ formatter = self.cell_formatters.get(row_name, {}).get(col_name) if not formatter: align_right = ( isinstance(value, (float, int)) or value is None or value == "?" ) value = self._format_value(value) formatter = CellFormatter(bold=bold, color=color, align_right=align_right) return formatter.format_value(value) def _get_markup(self, cells): """ Return a string cotaining all printable cells (see **_get_printable_column_order** and **_get_printable_row_order**) as correctly formatted markup. """ parts = [] for row_name in self._get_printable_row_order(): if row_name == self.header_row: parts.append(self._get_header_markup(row_name, cells[row_name])) else: parts.append(self._get_row_markup(row_name, cells[row_name])) if parts.append(" ".join( return "\n".join(parts) def _get_header_markup(self, row_name, row): """Return the txt2tags table markup for the headers.""" return self._get_row_markup(row_name, row, template="|| {} |") def _get_row_markup(self, row_name, row, template=" | {} |"): """Return the txt2tags table markup for one row.""" formatted_cells = [] for col_name in self._get_printable_column_order(): formatted_cells.append(row.get(col_name, "")) return template.format(" | ".join(formatted_cells)) def __str__(self): """Return the txt2tags markup for this table.""" cells = self._collect_cells() self._format(cells) return self._get_markup(cells) def extract_summary_rows(from_table, to_table, link=None): """ Extract all summary rows of **from_table** and add them as data rows to **to_table**. """ for name, row in from_table.get_summary_rows().items(): row_name = f"{from_table.title} - {name}" if link is not None: formatter = CellFormatter(link=link) to_table.cell_formatters[row_name][to_table.header_column] = formatter to_table.row_min_wins[row_name] = from_table.min_wins for col_name, value in row.items(): if col_name == from_table.header_column: continue to_table.add_cell(row_name, col_name, value) class DynamicDataModule: """Interface for modules that dynamically add or modify data in a table.""" def collect(self, table, cells): """ Called after the data collection in the table. Subclasses can add new values to **cells** or modify existing values. """ return cells def format(self, table, formatted_cells): """ Called after the formatting in the table. Subclasses can (re-)format all values in **formatted_cells**. Specifically all new values added by the **collect** method should be formatted. """ pass def modify_printable_row_order(self, table, row_order): """ Called after retrieving a row order in the table. Subclasses can modify the order or add new rows. Specifically all rows that were added by the **collect** method should be appended or inserted. """ return row_order def modify_printable_column_order(self, table, column_order): """ Called after retrieving a column order in the table. Subclassed can modify the order or add new columns. Specifically all columns that were values added by the **collect** method should be appended or inserted. """ return column_order