Merge pull request #167 from remia/feature/report_api_improvements

Report public API improvement and various fixes

clairmeta/dcp.py

@@ -293,9 +293,7 @@ def check(
                 ov_path (str, optional): Absolute path of OriginalVersion DCP.
 
             Returns:
-                Tuple (boolean, dict) of DCP check status and report. Report
-                    is a Dictionary where check failure are grouped by
-                    criticity.
+                Tuple (boolean, CheckReport) of DCP check status and report.
 
         """
         if not self._parsed or not self._probeb:

clairmeta/dcp_check.py

@@ -52,19 +52,19 @@ def load_modules(self):
             try:
                 module_path = 'clairmeta.' + prefix + k
                 module_vol = importlib.import_module(module_path)
-                checker = module_vol.Checker(self.dcp, self.check_profile)
+                checker = module_vol.Checker(self.dcp, self.profile)
                 checker.hash_callback = self.hash_callback
                 self.check_modules[v] = checker
             except (ImportError, Exception) as e:
-                self.check_log.critical("Import error {} : {}".format(
+                self.log.critical("Import error {} : {}".format(
                     module_path, str(e)))
 
     def check(self):
         """ Execute the complete check process. """
         self.run_checks()
         self.make_report()
         self.dump_report()
-        return self.get_valid(), self.check_report
+        return self.report.valid(), self.report
 
     def list_checks(self):
         """ List all available checks. """
@@ -92,7 +92,7 @@ def list_checks(self):
 
     def run_checks(self):
         """ Execute all checks. """
-        self.check_log.info("Checking DCP : {}".format(self.dcp.path))
+        self.log.info("Checking DCP : {}".format(self.dcp.path))
 
         # Run own tests
         dcp_checks = self.find_check('dcp')
@@ -101,7 +101,7 @@ def run_checks(self):
 
         # Run external modules tests
         for _, checker in six.iteritems(self.check_modules):
-            self.check_executions += checker.run_checks()
+            self.checks += checker.run_checks()
 
     def check_dcp_empty_dir(self):
         """ Empty directory detection.

clairmeta/dcp_check_am.py

@@ -28,7 +28,7 @@ def run_checks(self):
              for asset in list_am_assets(source)
              for check in asset_checks]
 
-        return self.check_executions
+        return self.checks
 
     def check_am_xml(self, am):
         """ AssetMap XML syntax and structure check.

clairmeta/dcp_check_atmos.py

@@ -23,10 +23,10 @@ def run_checks(self):
              if asset[1]['Schema'] == 'Atmos'
              for check in asset_checks]
 
-        return self.check_executions
+        return self.checks
 
     def check_atmos_cpl_essence_encoding(self, playlist, asset):
-        """ Atmos encoding.
+        """ Atmos data essence coding universal label.
 
             Reference :
                 SMPTE 429-18-2019 11

clairmeta/dcp_check_base.py

@@ -5,9 +5,10 @@
 import time
 import inspect
 import traceback
+from datetime import datetime
 
 from clairmeta.logger import get_log
-from clairmeta.utils.file import human_size
+from clairmeta.dcp_check_report import CheckReport
 
 
 class CheckException(Exception):
@@ -17,17 +18,29 @@ def __init__(self, msg):
 
 
 class CheckExecution(object):
-    """ Check execution with status and time elapsed. """
+    """ Check execution with status and related metadatas. """
 
-    def __init__(self, name):
-        self.name = name
-        self.doc = ""
+    def __init__(self, func):
+        """ Constructor for CheckExecution.
+
+            Args:
+                func (function): Check function.
+
+        """
+        self.name = func.__name__
+        self.doc = func.__doc__
         self.message = ""
         self.valid = False
+        self.bypass = False
         self.seconds_elapsed = 0
         self.asset_stack = []
         self.criticality = ""
 
+    def short_desc(self):
+        """ Returns first line of the docstring or function name. """
+        docstring_lines = self.doc.split('\n')
+        return docstring_lines[0].strip() if docstring_lines else c.name
+
 
 class CheckerBase(object):
     """ Base class for check module, provide check discover and run utilities.
@@ -45,10 +58,10 @@ def __init__(self, dcp, profile):
 
         """
         self.dcp = dcp
-        self.check_profile = profile
-        self.check_log = get_log()
-        self.check_executions = []
-        self.check_report = {}
+        self.profile = profile
+        self.log = get_log()
+        self.checks = []
+        self.report = None
         self.hash_callback = None
 
     def find_check_criticality(self, name):
@@ -61,7 +74,7 @@ def find_check_criticality(self, name):
                 Criticality level string.
 
         """
-        check_level = self.check_profile['criticality']
+        check_level = self.profile['criticality']
         default = check_level.get('default', 'ERROR')
         score_profile = {
             0: default
@@ -85,7 +98,7 @@ def find_check(self, prefix):
 
         """
         member_list = inspect.getmembers(self, predicate=inspect.ismethod)
-        bypass = self.check_profile['bypass']
+        bypass = self.profile['bypass']
 
         checks = []
         for k, v in member_list:
@@ -94,18 +107,18 @@ def find_check(self, prefix):
 
             if check_prefix and not check_bypass:
                 checks.append(v)
+            elif check_bypass:
+                check_exec = CheckExecution(v)
+                check_exec.bypass = True
+                self.checks.append(check_exec)
 
         return checks
 
-    def find_check_failed(self):
-        """ Returns a list of all failed checks. """
-        return [c for c in self.check_executions if not c.valid]
-
     def run_check(self, check, *args, **kwargs):
         """ Execute a check.
 
             Args:
-                check (tuple): Tuple (function name, function).
+                check (function): Check function.
                 *args: Variable list of check function arguments.
                 **kwargs: Variable list of keywords arguments.
                     error_prefix (str): error message prefix
@@ -114,171 +127,42 @@ def run_check(self, check, *args, **kwargs):
                 Tuple (status, return_value)
 
         """
-        start = time.time()
-        name, func = check.__name__, check
-        check_exec = CheckExecution(name)
-        check_exec.doc = check.__doc__
-        check_res = None
+        check_exec = CheckExecution(check)
+        check_exec.criticality = self.find_check_criticality(check_exec.name)
+        check_exec.valid = False
 
         try:
-            check_res = func(*args)
-            check_exec.valid = True
-            check_exec.msg = "Check valid"
+            start = time.time()
+            check_res = None
+            check_res = check(*args)
         except CheckException as e:
-            if kwargs.get('error_prefix'):
-                msg = "{}\n\t{}".format(kwargs.get('error_prefix'), str(e))
-            else:
-                msg = str(e)
-            check_exec.msg = msg
-            check_exec.criticality = self.find_check_criticality(name)
+            prefix = kwargs.get('error_prefix', '')
+            check_exec.message = "{}{}".format(
+                prefix + "\n\t" if prefix else '', str(e))
         except Exception as e:
-            check_exec.msg = "Check unknown error\n{}".format(
+            check_exec.message = "Check unknown error\n{}".format(
                 traceback.format_exc())
             check_exec.criticality = "ERROR"
-            self.check_log.error(check_exec.msg)
+            self.log.error(check.msg)
+        else:
+            check_exec.valid = True
         finally:
             check_exec.asset_stack = kwargs.get('stack', [self.dcp.path])
             check_exec.seconds_elapsed = time.time() - start
-            self.check_executions.append(check_exec)
+            self.checks.append(check_exec)
             return check_exec.valid, check_res
 
     def make_report(self):
         """ Check report generation. """
-        self.check_report = {
-            'ERROR': [],
-            'WARNING': [],
-            'INFO': [],
-            'SILENT': []
-        }
-
-        for c in self.find_check_failed():
-            self.check_report[c.criticality].append((c.name, c.msg))
-
-        check_unique = set([c.name for c in self.check_executions])
-        self.check_elapsed = {}
-        self.total_time = 0
-        self.total_check = len(check_unique)
+        report = CheckReport()
+        report.dcp = self.dcp
+        report.checks = self.checks
+        report.profile = self.profile
+        report.date = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
+        report.duration = sum([c.seconds_elapsed for c in self.checks])
 
-        for name in check_unique:
-            execs = [c.seconds_elapsed
-                     for c in self.check_executions if c.name == name]
-            elapsed = sum(execs)
-            self.total_time += elapsed
-            self.check_elapsed[name] = elapsed
+        self.report = report
 
     def dump_report(self):
         """ Dump check report. """
-        valid_str = 'Success' if self.get_valid() else 'Fail'
-
-        pretty_status = {
-            'ERROR': 'Error(s)',
-            'WARNING': 'Warning(s)',
-            'INFO': 'Info(s)',
-            'SILENT': 'Supressed(s)',
-            'BYPASS': 'Bypass(s)',
-        }
-
-        map_status = {
-            'ERROR': {},
-            'WARNING': {},
-            'INFO': {},
-            'SILENT': {},
-        }
-
-        # Accumulate all failed check and stack them by asset
-        for c in self.find_check_failed():
-            node = map_status[c.criticality]
-            for filename in c.asset_stack:
-                if filename not in node:
-                    node[filename] = {}
-
-                node = node[filename]
-                if 'messages' not in node:
-                    node['messages'] = []
-
-            docstring_lines = c.doc.split('\n')
-            desc = docstring_lines[0] if docstring_lines else c.name
-            node['messages'] += ['.' + desc + '\n' + c.msg]
-
-        self.check_log.info("DCP : {}".format(self.dcp.path))
-        self.check_log.info("Size : {}".format(human_size(self.dcp.size)))
-
-        for status, vals in six.iteritems(map_status):
-            out_stack = []
-            for k, v in six.iteritems(vals):
-                out_stack += [self.dump_stack("", k, v, indent_level=0)]
-            if out_stack:
-                self.check_log.info("{}\n{}".format(
-                    pretty_status[status] + ':',
-                    "\n".join(out_stack)))
-
-        if self.check_profile['bypass']:
-            checks_str = '  ' + '\n  '.join(self.check_profile['bypass'])
-            self.check_log.info("{}\n{}".format(
-                pretty_status['BYPASS'] + ':', checks_str))
-
-        self.check_log.info("Total check : {}".format(self.total_check))
-        self.check_log.info("Total time : {:.2f} sec".format(self.total_time))
-        self.check_log.info("Validation : {}\n".format(valid_str))
-
-    def dump_stack(self, out_str, key, values, indent_level):
-        """ Recursively iterate through the error message stack.
-
-            Args:
-                out_str (str): Accumulate messages to ``out_str``
-                key (str): Filename of the current asset.
-                values (dict): Message stack to dump.
-                indent_level (int): Current indentation level.
-
-            Returns:
-                Output error message string
-
-        """
-        indent_offset = 2
-        indent_step = 2
-        indent_char = ' '
-        ind = indent_offset + indent_level
-
-        filename = key
-        desc = self.title_from_filename(filename)
-        messages = values.pop('messages', [])
-
-        out_str = '' if indent_level == 0 else '\n'
-        out_str += indent_char * ind + '+ '
-        out_str += filename
-        out_str += ' ' + desc if desc else ''
-
-        ind += indent_step
-        for m in messages:
-            out_str += "\n"
-            out_str += indent_char * ind
-            # Correct indentation for multi-lines messages
-            out_str += ("\n" + indent_char * (ind + 2)).join(m.split("\n"))
-        ind -= indent_step
-
-        for k, v in six.iteritems(values):
-            out_str += self.dump_stack(
-                out_str, k, v, indent_level + indent_step)
-
-        return out_str
-
-    def title_from_filename(self, filename):
-        """ Returns a human friendly title for the given file. """
-        for cpl in self.dcp._list_cpl:
-            if cpl['FileName'] == filename:
-                desc = "({})".format(
-                    cpl['Info']['CompositionPlaylist'].get(
-                        'ContentTitleText', ''))
-                return desc
-
-        for pkl in self.dcp._list_pkl:
-            if pkl['FileName'] == filename:
-                desc = "({})".format(
-                    pkl['Info']['PackingList'].get('AnnotationText', ''))
-                return desc
-
-        return ''
-
-    def get_valid(self):
-        """ Check status is valid. """
-        return self.check_report['ERROR'] == []
+        self.log.info("Check report:\n\n" + self.report.pretty_str())

clairmeta/dcp_check_cpl.py

@@ -37,7 +37,7 @@ def run_checks(self):
              for asset in list_cpl_assets(source)
              for check in asset_checks]
 
-        return self.check_executions
+        return self.checks
 
     def metadata_cmp_pair(
         self,