(entrypoints: EntryPoints)
| 21 | |
| 22 | |
| 23 | def remove_duplicates(entrypoints: EntryPoints) -> list[EntryPoint]: |
| 24 | # sort and group entrypoints by name |
| 25 | entrypoints_sorted = sorted(entrypoints, key=lambda ep: ep.name) |
| 26 | entrypoints_grouped = itertools.groupby(entrypoints_sorted, key=lambda ep: ep.name) |
| 27 | # check if there are multiple entrypoints for the same name |
| 28 | unique_entrypoints = [] |
| 29 | for name, _matches in entrypoints_grouped: |
| 30 | # remove equal entrypoints |
| 31 | matches = list(set(_matches)) |
| 32 | unique_entrypoints.append(matches[0]) |
| 33 | matches_len = len(matches) |
| 34 | if matches_len > 1: |
| 35 | all_module_names = [e.value.split(":")[0] for e in matches] |
| 36 | selected_module_name = all_module_names[0] |
| 37 | warnings.warn( |
| 38 | f"Found {matches_len} entrypoints for the engine name {name}:" |
| 39 | f"\n {all_module_names}.\n " |
| 40 | f"The entrypoint {selected_module_name} will be used.", |
| 41 | RuntimeWarning, |
| 42 | stacklevel=2, |
| 43 | ) |
| 44 | return unique_entrypoints |
| 45 | |
| 46 | |
| 47 | def detect_parameters(open_dataset: Callable) -> tuple[str, ...]: |
no test coverage detected
searching dependent graphs…