(
path: Optional[str],
tags_yaml_path: str,
function_filter: Callable[[NativeFunction], bool],
skip_native_fns_gen: bool = False,
)
| 676 | |
| 677 | |
| 678 | def parse_yaml( |
| 679 | path: Optional[str], |
| 680 | tags_yaml_path: str, |
| 681 | function_filter: Callable[[NativeFunction], bool], |
| 682 | skip_native_fns_gen: bool = False, |
| 683 | ) -> Tuple[ |
| 684 | List[NativeFunction], |
| 685 | Union[Dict[DispatchKey, Dict[OperatorName, BackendMetadata]], ETKernelIndex], |
| 686 | ]: |
| 687 | if path and os.path.exists(path) and os.stat(path).st_size > 0: |
| 688 | with open(path) as f: |
| 689 | es = yaml.load(f, Loader=LineLoader) |
| 690 | |
| 691 | # Check for kernel index structure |
| 692 | kernel_index = ( |
| 693 | parse_et_yaml_struct(es) if any("kernels" in e for e in es) else None |
| 694 | ) |
| 695 | |
| 696 | # Remove ET specific fields from entries for BC compatibility |
| 697 | for entry in es: |
| 698 | for field in ET_FIELDS: |
| 699 | entry.pop(field, None) |
| 700 | |
| 701 | parsed_yaml = parse_native_yaml( |
| 702 | path, |
| 703 | tags_yaml_path, |
| 704 | None, |
| 705 | skip_native_fns_gen=skip_native_fns_gen, |
| 706 | loaded_yaml=es, |
| 707 | ) |
| 708 | native_functions = list(filter(function_filter, parsed_yaml.native_functions)) |
| 709 | op_names = [f.func.name for f in native_functions] |
| 710 | |
| 711 | # (1) Return ETKernelIndex if kernel index is present |
| 712 | if kernel_index is not None: |
| 713 | filtered_index = { |
| 714 | op_name: kernel_mapping |
| 715 | for op_name, kernel_mapping in kernel_index.index.items() |
| 716 | if op_name in op_names |
| 717 | } |
| 718 | return native_functions, ETKernelIndex(index=filtered_index) |
| 719 | |
| 720 | # (2) Return BackendIndices if kernel index is absent |
| 721 | def map_index( |
| 722 | m: Dict[OperatorName, BackendMetadata] |
| 723 | ) -> Dict[OperatorName, BackendMetadata]: |
| 724 | return {op: m[op] for op in m if op in op_names} |
| 725 | |
| 726 | backend_indices = { |
| 727 | k: map_index(b.index) for (k, b) in parsed_yaml.backend_indices.items() |
| 728 | } |
| 729 | |
| 730 | return native_functions, backend_indices |
| 731 | else: |
| 732 | return [], {} |
| 733 | |
| 734 | |
| 735 | def parse_yaml_files( |
no test coverage detected
searching dependent graphs…