Source code for benchbuild.reports
"""
Register reports for an experiment
"""
import importlib
import logging
import typing as t
import uuid
import attr
from benchbuild.settings import CFG
from benchbuild.utils import schema
LOG = logging.getLogger(__name__)
[docs]def discover():
"""
Import all experiments listed in *_PLUGINS_REPORTS.
Tests:
>>> from benchbuild.settings import CFG
>>> from benchbuild.reports import discover
>>> import logging as lg
>>> import sys
>>> l = lg.getLogger('benchbuild')
>>> l.setLevel(lg.DEBUG)
>>> l.handlers = [lg.StreamHandler(stream=sys.stdout)]
>>> CFG["plugins"]["reports"] = ["benchbuild.non.existing", "benchbuild.reports.raw"]
>>> discover()
Could not find 'benchbuild.non.existing'
Found report: benchbuild.reports.raw
"""
if CFG["plugins"]["autoload"]:
report_plugins = CFG["plugins"]["reports"].value
for plugin in report_plugins:
try:
importlib.import_module(plugin)
LOG.debug("Found report: %s", plugin)
except ImportError:
LOG.error("Could not find '%s'", plugin)
[docs]class ReportRegistry(type):
reports = {}
def __init__(cls, name, bases, _dict):
super(ReportRegistry, cls).__init__(name, bases, _dict)
if cls.NAME is not None:
ReportRegistry.reports[cls.NAME] = cls
[docs]def load_experiment_ids_from_names(session, names):
from sqlalchemy import func, column
from sqlalchemy.sql import select, bindparam
exps = select([column('id')]).\
select_from(
func.experiments(bindparam('names'))
)
r1 = session.execute(exps.unique_params(names=names))
return r1.fetchall()
[docs]@attr.s
class Report(metaclass=ReportRegistry):
SUPPORTED_EXPERIMENTS = []
NAME = None
def __new__(cls, *args, **kwargs):
del args, kwargs # Temporarily unused
new_self = super(Report, cls).__new__(cls)
if not cls.SUPPORTED_EXPERIMENTS:
raise AttributeError(
"{0} @ {1} does not define a SUPPORTED_EXPERIMENTS attribute"
.format(cls.__name__, cls.__module__))
if cls.NAME is None:
raise AttributeError("{0} @ {1} does not define a NAME attribute"
.format(cls.__name__, cls.__module__))
new_self.name = cls.NAME
return new_self
experiment_name = attr.ib()
exp_ids = attr.ib()
out_path = attr.ib()
session = attr.ib()
name = attr.ib(
default=attr.Factory(lambda self: type(self).NAME, takes_self=True))
supported_experiments = attr.ib(
default=attr.Factory(lambda self: type(self).NAME, takes_self=True))
experiment_ids = attr.ib(default=None)
def __attrs_post_init__(self):
if not self.exp_ids:
exp_ids = load_experiment_ids_from_names(self.session, [
exp for exp in self.SUPPORTED_EXPERIMENTS
if exp == self.experiment_name
])
exp_ids = [v[0] for v in exp_ids]
else:
exp_ids = [uuid.UUID(v) for v in self.exp_ids]
self.experiment_ids = exp_ids