Source code for fontbakery.reporters

Separation of Concerns Disclaimer:
While created specifically for checking fonts and font-families this
module has no domain knowledge about fonts. It can be used for any kind
of (document) checking. Please keep it so. It will be valuable for other
domains as well.
Domain specific knowledge should be encoded only in the Profile (Checks,
Conditions) and MAYBE in *customized* reporters e.g. subclasses.
from collections import Counter

from fontbakery.status import END, ENDCHECK, START
from fontbakery.errors import ProtocolViolationError

[docs]class FontbakeryReporter: def __init__(self, is_async=False, runner=None, output_file=None, loglevels=None): self._started = None self._ended = None self._order = None self._results = [] # ENDCHECK events in order of appearance self._indexes = {} self._tick = 0 self._counter = Counter() self.loglevels = loglevels # Runner should know if it is async! self.is_async = is_async self.runner = runner self._worst_check_status = None self.output_file = output_file
[docs] def run(self, order=None): """ self.runner must be present """ for event in self.receive(event)
@property def order(self): return self._order
[docs] def write(self): if self.output_file is not None: raise NotImplementedError( f'{type(self)} does not implement the "write" method, ' 'but it has an "output_file".' )
# reporters without an output file do nothing here @staticmethod def _get_key(identity): section, check, iterargs = identity return ( str(section) if section else section, str(check) if check else check, iterargs, ) def _get_index(self, identity): key = self._get_key(identity) try: return self._indexes[key] except KeyError: self._indexes[key] = len(self._indexes) return self._indexes[key] def _set_order(self, order): self._order = tuple(order) length = len(self._order) self._counter["(not finished)"] = length - len(self._results) self._indexes = dict(zip(map(self._get_key, self._order), range(length))) def _cleanup(self, event): pass def _output(self, event): pass def _register(self, event): status, message, identity = event self._tick += 1 if status == START: self._set_order(message) self._started = event if status == END: self._ended = event if status == ENDCHECK: self._results.append(event) self._counter[] += 1 self._counter["(not finished)"] -= 1 @property def worst_check_status(self): """Returns a status or None if there was no check result""" return self._worst_check_status
[docs] def receive(self, event): status, message, identity = event if self._started is None and status != START: raise ProtocolViolationError( f"Received Event before status START:" f" {status} {message}." ) if self._ended: status, message, identity = event raise ProtocolViolationError( f"Received Event after status END:" f" {status} {message}." ) if status is ENDCHECK and ( self._worst_check_status is None or self._worst_check_status < message ): # Checks that are marked as "experimental" do not affect the # exit status code, so that they won't break a build on continuous # integration setups. section, check, iterargs = identity if not check.experimental: # we only record ENDCHECK, because check runner may in the future # have tools to upgrade/downgrade the actually worst status # this should be future proof. self._worst_check_status = message self._register(event) self._cleanup(event) self._output(event)