Commit 1018e059 authored by Mitar's avatar Mitar

Added a script to validate which primitives are missing pipelines.

parent 79207e11
#!/usr/bin/env python3
import argparse
import glob
import json
parser = argparse.ArgumentParser(description="Run pipeline validation.")
parser.add_argument('interface_versions', metavar='INTERFACE', nargs='*', help="interface versions to validate", default=())
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))
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'])
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment