validate_coverage.py 2.11 KB
Newer Older
1 2 3 4 5 6
#!/usr/bin/env python3

import argparse
import glob
import json

Mitar's avatar
Mitar committed
7 8
import yaml

Mitar's avatar
Mitar committed
9 10
parser = argparse.ArgumentParser(description="Compute pipeline coverage.")
parser.add_argument('interface_versions', metavar='INTERFACE', nargs='*', help="interface version(s) to compute coverage for", default=())
11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43
arguments = parser.parse_args()


def primitives_used_in_pipeline(pipeline):
    primitives = set()

    for step in pipeline.get('steps', []):
        if step['type'] == 'PRIMITIVE':
            primitives.add(step['primitive']['id'])
        elif step['type'] == 'SUBPIPELINE':
            primitives.update(primitives_used_in_pipeline(step['pipeline']))
        else:
            raise ValueError(f"Unsupported pipeline step type: {step['type']}")

    return primitives


for interface_version in arguments.interface_versions:
    known_primitives = {}
    primitives_used_in_pipelines = set()

    for primitive_annotation_path in glob.iglob('{interface_version}/*/*/*/primitive.json'.format(interface_version=interface_version)):
        with open(primitive_annotation_path, 'r', encoding='utf8') as primitive_annotation_file:
            primitive_annotation = json.load(primitive_annotation_file)

        known_primitives[primitive_annotation['id']] = primitive_annotation

    for pipeline_path in glob.iglob('{interface_version}/*/*/*/pipelines/*.json'.format(interface_version=interface_version)):
        with open(pipeline_path, 'r', encoding='utf8') as pipeline_file:
            pipeline = json.load(pipeline_file)

        primitives_used_in_pipelines.update(primitives_used_in_pipeline(pipeline))

Mitar's avatar
Mitar committed
44 45 46 47 48 49
    for pipeline_path in glob.iglob('{interface_version}/*/*/*/pipelines/*.yml'.format(interface_version=interface_version)):
        with open(pipeline_path, 'r', encoding='utf8') as pipeline_file:
            pipeline = yaml.safe_load(pipeline_file)

        primitives_used_in_pipelines.update(primitives_used_in_pipeline(pipeline))

50 51 52
    for primitive_id, primitive in known_primitives.items():
        if primitive_id not in primitives_used_in_pipelines:
            print(primitive['source']['name'], primitive['id'], primitive['python_path'])