Skip to content
Snippets Groups Projects
Select Git revision
  • bc0ce88cf9d366d1ddc5e82800c12297ddb005c0
  • master default protected
  • MAM-90-executor-with-cwltool
  • MAM-88-change-status-names
  • MAM-50-specification-GUI
  • MAM-44-build-specification-functionality
  • MAM-42-ATDBspec-add-endpoints
  • MAM-41-ATDBspec-database-changes
  • bugfix-ingested-sizes
  • SDC-1663-inputs-validation
  • SDC-1649-compression-pipeline-quality-indicators
  • SDC-1635-monitoring-page
  • split-pipeline-for-gitlab
  • SDC-1590-database-changes
  • SDC-1580-adapt-configuration-to-multi-service
  • multi-edit
  • SDC-1552-datamodel-add-service-host
  • logentry_graph
  • SDC-1549-aggregation-task-ux
  • imaging-compression-pipeline-commissioning
  • unit-tests
21 results

common.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    common.py 1.92 KiB
    """
    common helper functions
    """
    import logging;
    from datetime import *
    import time
    from enum import Enum
    
    logger = logging.getLogger(__name__)
    
    class SummaryFlavour(Enum):
        DEFAULT = "default"
        LINC_CALIBRATOR = "linc_calibrator"
        LINC_TARGET = "linc_target"
        IMAGING_COMPRESSION = "imaging_compression"
    
    # this is a decorator that can be put in front (around) a function all to measure its execution time
    def timeit(method):
        def timed(*args, **kw):
            ts = time.time()
            result = method(*args, **kw)
            te = time.time()
            if 'log_time' in kw:
                name = kw.get('log_name', method.__name__.upper())
                kw['log_time'][name] = int((te - ts) * 1000)
            else:
                print('execution time: %r  %2.2f ms' % \
                      (method.__name__, (te - ts) * 1000))
            return result
        return timed
    
    def get_summary_flavour(task):
        """
        # not every workflow has the same summary structure
        # determine the flavour based on the selected task, and construct the html accordingly
        # this could be made implicit in the future by adding a setting to the Workflow,
        # but currently it is derived in a yucky way. But at least the yuck is confined to this function
        """
        summary_flavour = SummaryFlavour.DEFAULT.value
    
        try:
            summary = task.quality_json["summary"]
        except:
            # no summary found
            return None
    
        # so... yikes... what distinguishes the summary flavours?
        workflow_uri = task.workflow.workflow_uri
    
        # for linc, look at the workflow_uri
        if "linc_calibrator" in workflow_uri:
            return SummaryFlavour.LINC_CALIBRATOR.value
        if "linc_target" in workflow_uri:
            return SummaryFlavour.LINC_TARGET.value
    
        try:
            d = summary["details"]
            summary_flavour = SummaryFlavour.IMAGING_COMPRESSION.value
        except:
            # this is not an imaging summary, continue with the default
            pass
    
        return summary_flavour