Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/utils/optype_collector/optype_collector/optype_collector_main.py: 98%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------------------------------------- 

4# Copyright (c) 2026 Huawei Technologies Co., Ltd. 

5# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

6# CANN Open Software License Agreement Version 2.0 (the "License"). 

7# Please refer to the License for details. You may not use this file except in compliance with the License. 

8# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

9# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

10# See LICENSE in the root of the software repository for the full text of the License. 

11# ---------------------------------------------------------------------------------------------------------- 

12"""Collect and detect duplicated OpType definitions in CANN OPP packages.""" 

13 

14import argparse 

15import json 

16import os 

17import sys 

18from dataclasses import dataclass, field 

19from pathlib import Path 

20from typing import Any, Dict, Iterable, List, Optional, Sequence, Set, Tuple 

21 

22BUILTIN = "builtin" 

23CUSTOM = "custom" 

24CUSTOM_OPP_ENV = "ASCEND_CUSTOM_OPP_PATH" 

25ASCEND_HOME_ENV = "ASCEND_HOME_PATH" 

26CANN_ENV_ERROR = "Error: CANN environment variables are not configured." 

27CANN_ENV_SOURCE_HINT = "Please execute: source <CANN_install_path>/cann/set_env.sh" 

28 

29 

30@dataclass 

31class OpTypeSource: 

32 """One OpType scan source.""" 

33 

34 source_type: str 

35 soc: str 

36 matched_soc: str 

37 root_path: Path 

38 vendor_name: Optional[str] = None 

39 config_files: List[Path] = field(default_factory=list) 

40 optypes: Set[str] = field(default_factory=set) 

41 warnings: List[str] = field(default_factory=list) 

42 errors: List[str] = field(default_factory=list) 

43 

44 @property 

45 def status(self) -> str: 

46 if self.errors: 

47 return "ERROR" 

48 if self.warnings: 

49 return "WARN" 

50 return "OK" 

51 

52 @property 

53 def identity(self) -> Tuple[str, str, str, str]: 

54 return ( 

55 self.source_type, 

56 self.vendor_name or "", 

57 self.matched_soc, 

58 str(self.root_path), 

59 ) 

60 

61 

62@dataclass 

63class ScanResult: 

64 """Full scan result.""" 

65 

66 user_soc: str 

67 soc_names: List[str] 

68 ascend_home_path: Optional[Path] 

69 custom_opp_path: Optional[str] 

70 builtin_sources: List[OpTypeSource] = field(default_factory=list) 

71 custom_sources: List[OpTypeSource] = field(default_factory=list) 

72 warnings: List[str] = field(default_factory=list) 

73 errors: List[str] = field(default_factory=list) 

74 

75 

76@dataclass 

77class ConflictGroup: 

78 """One duplicated OpType conflict group.""" 

79 

80 conflict_type: str 

81 left: OpTypeSource 

82 right: OpTypeSource 

83 optypes: List[str] 

84 

85 

86@dataclass 

87class ConflictReport: 

88 """Conflict detection report.""" 

89 

90 custom_builtin: List[ConflictGroup] = field(default_factory=list) 

91 custom_custom: List[ConflictGroup] = field(default_factory=list) 

92 

93 @property 

94 def has_conflicts(self) -> bool: 

95 return bool(self.custom_builtin or self.custom_custom) 

96 

97 

98@dataclass 

99class SocNameMap: 

100 """Map public SoC versions to internal OPP config directory names.""" 

101 

102 external_to_short: Dict[str, List[str]] = field(default_factory=dict) 

103 external_lower_to_short: Dict[str, List[str]] = field(default_factory=dict) 

104 short_to_external: Dict[str, List[str]] = field(default_factory=dict) 

105 

106 

107def _append_unique(items: List[str], item: str) -> None: 

108 if item and item not in items: 

109 items.append(item) 

110 

111 

112def _read_platform_config_soc_names( 

113 config_file: Path, 

114) -> Tuple[Optional[str], Optional[str]]: 

115 external_soc = None 

116 short_soc = None 

117 try: 

118 with config_file.open("r", encoding="utf-8") as file_obj: 

119 for raw_line in file_obj: 

120 line = raw_line.strip() 

121 if not line or line.startswith("#") or "=" not in line: 

122 continue 

123 key, value = [part.strip() for part in line.split("=", 1)] 

124 lowered_key = key.lower() 

125 if lowered_key == "soc_version": 

126 external_soc = value 

127 elif lowered_key == "short_soc_version": 

128 short_soc = value 

129 except OSError: 

130 return None, None 

131 return external_soc, short_soc 

132 

133 

134def _platform_config_dirs(ascend_home_path: Path) -> List[Path]: 

135 """Return platform_config directories without assuming a fixed CPU architecture.""" 

136 candidates = [] 

137 for child in sorted(ascend_home_path.glob("*-linux")): 

138 platform_config_dir = child / "data" / "platform_config" 

139 if platform_config_dir.is_dir(): 

140 candidates.append(platform_config_dir) 

141 return candidates 

142 

143 

144def load_soc_name_map(ascend_home_path: Optional[Path]) -> SocNameMap: 

145 """Load public SoC name and internal short-name mapping from CANN platform_config.""" 

146 name_map = SocNameMap() 

147 if not ascend_home_path: 

148 return name_map 

149 for platform_config_dir in _platform_config_dirs(ascend_home_path): 

150 for config_file in sorted(platform_config_dir.glob("*.ini")): 

151 external_soc, short_soc = _read_platform_config_soc_names(config_file) 

152 if not external_soc or not short_soc: 

153 continue 

154 short_key = short_soc.lower() 

155 _append_unique( 

156 name_map.external_to_short.setdefault(external_soc, []), short_key 

157 ) 

158 _append_unique( 

159 name_map.external_lower_to_short.setdefault(external_soc.lower(), []), 

160 short_key, 

161 ) 

162 _append_unique( 

163 name_map.short_to_external.setdefault(short_key, []), external_soc 

164 ) 

165 return name_map 

166 

167 

168def _expand_internal_soc_names(soc_names: Iterable[str]) -> List[str]: 

169 alias_map = { 

170 "ascend950": ["ascend950", "ascend910_95"], 

171 } 

172 expanded_names = [] 

173 for soc_name in soc_names: 

174 for expanded_name in alias_map.get(soc_name, [soc_name]): 

175 _append_unique(expanded_names, expanded_name) 

176 return expanded_names 

177 

178 

179def expand_soc_aliases( 

180 soc_version: str, soc_name_map: Optional[SocNameMap] = None 

181) -> List[str]: 

182 """Resolve SoC input to OPP config directory names.""" 

183 normalized_soc = soc_version.lower() 

184 if soc_version == normalized_soc: 

185 return _expand_internal_soc_names([normalized_soc]) 

186 if soc_name_map: 

187 mapped_names = soc_name_map.external_to_short.get(soc_version) 

188 if not mapped_names: 

189 mapped_names = soc_name_map.external_lower_to_short.get(normalized_soc) 

190 if mapped_names: 

191 return _expand_internal_soc_names(mapped_names) 

192 return [] 

193 

194 

195def _read_json_file(json_file: Path, source: OpTypeSource) -> Optional[Any]: 

196 try: 

197 with json_file.open("r", encoding="utf-8") as file_obj: 

198 return json.load(file_obj) 

199 except Exception as err: 

200 source.warnings.append( 

201 "Failed to parse JSON: {}; reason: {}".format(json_file, err) 

202 ) 

203 return None 

204 

205 

206def _looks_like_op_type(name: str) -> bool: 

207 if not name or name.startswith("_"): 

208 return False 

209 lowered = name.lower() 

210 non_op_keys = { 

211 "bininfo", 

212 "bin_info", 

213 "supportinfo", 

214 "support_info", 

215 "opinfo", 

216 "op_info", 

217 "oplist", 

218 "op_list", 

219 "ops", 

220 "op", 

221 "socversion", 

222 "soc_version", 

223 "version", 

224 "platform", 

225 "impl_path", 

226 "dynamic_compile_static", 

227 "static_compile", 

228 "dynamic_compile", 

229 "simplifiedkeymode", 

230 "simplified_key_mode", 

231 "computeunit", 

232 "compute_unit", 

233 "opfile", 

234 "op_file", 

235 "jsonfilepath", 

236 "json_file_path", 

237 "kernelname", 

238 "kernel_name", 

239 "input", 

240 "output", 

241 "attr", 

242 "attrs", 

243 "dtype", 

244 "format", 

245 "precision_reduce", 

246 "enable_vector_core", 

247 } 

248 return lowered not in non_op_keys 

249 

250 

251def _collect_optypes_from_json(data: Any) -> Set[str]: 

252 optypes = set() 

253 container_keys = { 

254 "ops", 

255 "opList", 

256 "op_list", 

257 "op_info", 

258 "opInfo", 

259 "binInfo", 

260 "bin_info", 

261 } 

262 explicit_optype_keys = { 

263 "opType", 

264 "op_type", 

265 "opTypeName", 

266 "op_type_name", 

267 "opName", 

268 "op_name", 

269 } 

270 

271 def visit(obj: Any, parent_key: Optional[str] = None) -> None: 

272 if isinstance(obj, dict): 

273 for key, value in obj.items(): 

274 if key in explicit_optype_keys: 

275 if isinstance(value, str) and value: 

276 optypes.add(value) 

277 continue 

278 if key == "name" and parent_key in container_keys: 

279 if isinstance(value, str) and value: 

280 optypes.add(value) 

281 continue 

282 

283 key_is_optype = ( 

284 isinstance(key, str) 

285 and isinstance(value, (dict, list)) 

286 and (parent_key is None or parent_key in container_keys) 

287 and _looks_like_op_type(key) 

288 ) 

289 if key_is_optype: 

290 optypes.add(key) 

291 visit(value, key) 

292 elif isinstance(value, (dict, list)): 

293 visit(value, key) 

294 elif isinstance(obj, list): 

295 for item in obj: 

296 visit(item, parent_key) 

297 

298 visit(data) 

299 return optypes 

300 

301 

302def _scan_config_dir( 

303 source_type: str, 

304 soc: str, 

305 matched_soc: str, 

306 root_path: Path, 

307 vendor_name: Optional[str] = None, 

308) -> OpTypeSource: 

309 source = OpTypeSource( 

310 source_type=source_type, 

311 soc=soc, 

312 matched_soc=matched_soc, 

313 vendor_name=vendor_name, 

314 root_path=root_path, 

315 ) 

316 if not root_path.exists(): 

317 source.errors.append("Path does not exist: {}".format(root_path)) 

318 return source 

319 if not root_path.is_dir(): 

320 source.errors.append("Scan path is not a directory: {}".format(root_path)) 

321 return source 

322 

323 json_files = sorted(root_path.rglob("*.json")) 

324 source.config_files = json_files 

325 if not json_files: 

326 source.warnings.append("No JSON config files found: {}".format(root_path)) 

327 return source 

328 

329 for json_file in json_files: 

330 data = _read_json_file(json_file, source) 

331 if data is None: 

332 continue 

333 source.optypes.update(_collect_optypes_from_json(data)) 

334 if not source.optypes: 

335 source.warnings.append("No OpTypes parsed from: {}".format(root_path)) 

336 return source 

337 

338 

339def _source_merge_key(source: OpTypeSource) -> Tuple[str, str, str]: 

340 """Return package identity, ignoring alias SoC directory names under the same package.""" 

341 return (source.source_type, source.vendor_name or "", str(source.root_path.parent)) 

342 

343 

344def _merge_matched_soc(left: str, right: str) -> str: 

345 soc_names = [] 

346 for item in (left, right): 

347 for soc_name in item.split(","): 

348 soc_name = soc_name.strip() 

349 if soc_name and soc_name not in soc_names: 

350 soc_names.append(soc_name) 

351 return ",".join(soc_names) 

352 

353 

354def _merge_config_files(left: List[Path], right: Iterable[Path]) -> List[Path]: 

355 """Merge config file lists while preserving order and removing duplicate paths.""" 

356 merged_files = list(left) 

357 seen = {str(path) for path in merged_files} 

358 for path in right: 

359 path_key = str(path) 

360 if path_key not in seen: 

361 merged_files.append(path) 

362 seen.add(path_key) 

363 return merged_files 

364 

365 

366def _deduplicate_sources(sources: Iterable[OpTypeSource]) -> List[OpTypeSource]: 

367 merged: Dict[Tuple[str, str, str], OpTypeSource] = {} 

368 for source in sources: 

369 key = _source_merge_key(source) 

370 if key not in merged: 

371 source.config_files = _merge_config_files([], source.config_files) 

372 merged[key] = source 

373 continue 

374 exist = merged[key] 

375 exist.matched_soc = _merge_matched_soc(exist.matched_soc, source.matched_soc) 

376 exist.config_files = _merge_config_files( 

377 exist.config_files, source.config_files 

378 ) 

379 exist.optypes.update(source.optypes) 

380 exist.warnings.extend(source.warnings) 

381 exist.errors.extend(source.errors) 

382 return list(merged.values()) 

383 

384 

385def _candidate_builtin_dirs( 

386 ascend_home_path: Path, soc_names: Sequence[str] 

387) -> List[Tuple[str, Path]]: 

388 base = ( 

389 ascend_home_path / "opp" / "built-in" / "op_impl" / "ai_core" / "tbe" / "config" 

390 ) 

391 candidates = [] 

392 for soc_name in soc_names: 

393 candidates.append((soc_name, base / soc_name)) 

394 if base.exists(): 

395 for child in sorted(base.iterdir()): 

396 if child.is_dir() and child.name != soc_name: 

397 candidates.append((soc_name, child / soc_name)) 

398 return candidates 

399 

400 

401def collect_builtin_optypes( 

402 ascend_home_path: Path, user_soc: str, soc_names: Sequence[str] 

403) -> List[OpTypeSource]: 

404 sources = [] 

405 for matched_soc, path in _candidate_builtin_dirs(ascend_home_path, soc_names): 

406 if path.exists(): 

407 sources.append(_scan_config_dir(BUILTIN, user_soc, matched_soc, path)) 

408 return _deduplicate_sources(sources) 

409 

410 

411def _vendor_dirs( 

412 vendors_root: Path, soc_names: Sequence[str] 

413) -> List[Tuple[str, str, Path]]: 

414 candidates = [] 

415 if not vendors_root.exists(): 

416 return candidates 

417 for vendor_dir in sorted(vendors_root.iterdir()): 

418 if not vendor_dir.is_dir(): 

419 continue 

420 for soc_name in soc_names: 

421 candidates.append( 

422 ( 

423 vendor_dir.name, 

424 soc_name, 

425 vendor_dir / "op_impl" / "ai_core" / "tbe" / "config" / soc_name, 

426 ) 

427 ) 

428 return candidates 

429 

430 

431def _custom_opp_dirs( 

432 custom_root: Path, soc_names: Sequence[str] 

433) -> List[Tuple[str, str, Path]]: 

434 candidates = [] 

435 for soc_name in soc_names: 

436 direct_path = custom_root / "op_impl" / "ai_core" / "tbe" / "config" / soc_name 

437 candidates.append((custom_root.name, soc_name, direct_path)) 

438 if not custom_root.exists(): 

439 return candidates 

440 for vendor_dir in sorted(custom_root.iterdir()): 

441 if not vendor_dir.is_dir(): 

442 continue 

443 for soc_name in soc_names: 

444 candidates.append( 

445 ( 

446 vendor_dir.name, 

447 soc_name, 

448 vendor_dir / "op_impl" / "ai_core" / "tbe" / "config" / soc_name, 

449 ) 

450 ) 

451 return candidates 

452 

453 

454def collect_custom_optypes( 

455 ascend_home_path: Path, 

456 custom_opp_path: Optional[str], 

457 user_soc: str, 

458 soc_names: Sequence[str], 

459) -> Tuple[List[OpTypeSource], List[str]]: 

460 warnings = [] 

461 sources = [] 

462 vendors_root = ascend_home_path / "opp" / "vendors" 

463 if not vendors_root.exists(): 

464 warnings.append( 

465 "${{ASCEND_HOME_PATH}}/opp/vendors not found: {}".format(vendors_root) 

466 ) 

467 for vendor, matched_soc, path in _vendor_dirs(vendors_root, soc_names): 

468 if path.exists(): 

469 sources.append( 

470 _scan_config_dir(CUSTOM, user_soc, matched_soc, path, vendor) 

471 ) 

472 

473 if not custom_opp_path: 

474 warnings.append( 

475 "ASCEND_CUSTOM_OPP_PATH is not set; " 

476 "only custom packages under ${ASCEND_HOME_PATH}/opp/vendors will be scanned." 

477 ) 

478 return _deduplicate_sources(sources), warnings 

479 

480 for custom_root_str in custom_opp_path.split(os.pathsep): 

481 if not custom_root_str: 

482 continue 

483 custom_root = Path(custom_root_str) 

484 if not custom_root.exists(): 

485 warnings.append( 

486 "ASCEND_CUSTOM_OPP_PATH Path does not exist: {}".format(custom_root) 

487 ) 

488 for vendor, matched_soc, path in _custom_opp_dirs(custom_root, soc_names): 

489 if path.exists(): 

490 sources.append( 

491 _scan_config_dir(CUSTOM, user_soc, matched_soc, path, vendor) 

492 ) 

493 return _deduplicate_sources(sources), warnings 

494 

495 

496def _available_soc_dir_names( 

497 ascend_home_path: Path, custom_opp_path: Optional[str] = None 

498) -> List[str]: 

499 roots = [ 

500 ascend_home_path 

501 / "opp" 

502 / "built-in" 

503 / "op_impl" 

504 / "ai_core" 

505 / "tbe" 

506 / "config", 

507 ascend_home_path / "opp" / "vendors", 

508 ] 

509 if custom_opp_path: 

510 roots.extend(Path(path) for path in custom_opp_path.split(os.pathsep) if path) 

511 socs = set() 

512 for root in roots: 

513 if not root.exists(): 

514 continue 

515 for path in root.rglob("*"): 

516 if path.is_dir() and path.name.lower().startswith("ascend"): 

517 socs.add(path.name) 

518 return sorted(socs) 

519 

520 

521def scan_optypes( 

522 user_soc: str, need_builtin: bool = True, need_custom: bool = True 

523) -> ScanResult: 

524 ascend_home = os.environ.get(ASCEND_HOME_ENV) 

525 custom_opp_path = os.environ.get(CUSTOM_OPP_ENV) 

526 ascend_home_path = Path(ascend_home) if ascend_home else None 

527 soc_name_map = load_soc_name_map(ascend_home_path) 

528 soc_names = expand_soc_aliases(user_soc, soc_name_map) 

529 result = ScanResult( 

530 user_soc=user_soc, 

531 soc_names=soc_names, 

532 ascend_home_path=ascend_home_path, 

533 custom_opp_path=custom_opp_path, 

534 ) 

535 if not ascend_home_path: 

536 result.errors.append(CANN_ENV_ERROR) 

537 result.errors.append(CANN_ENV_SOURCE_HINT) 

538 return result 

539 if not (ascend_home_path / "opp").exists(): 

540 result.errors.append(CANN_ENV_ERROR) 

541 result.errors.append(CANN_ENV_SOURCE_HINT) 

542 return result 

543 if not soc_names: 

544 result.errors.append(_format_unsupported_soc_error(result.user_soc)) 

545 return result 

546 

547 if need_builtin: 

548 result.builtin_sources = collect_builtin_optypes( 

549 ascend_home_path, result.user_soc, soc_names 

550 ) 

551 if not result.builtin_sources: 

552 result.errors.append(_format_unsupported_soc_error(result.user_soc)) 

553 if need_custom: 

554 result.custom_sources, custom_warnings = collect_custom_optypes( 

555 ascend_home_path, custom_opp_path, result.user_soc, soc_names 

556 ) 

557 result.warnings.extend(custom_warnings) 

558 if not result.custom_sources: 

559 available_soc_dir_names = _available_soc_dir_names( 

560 ascend_home_path, custom_opp_path 

561 ) 

562 if available_soc_dir_names and not set(soc_names).intersection( 

563 {soc.lower() for soc in available_soc_dir_names} 

564 ): 

565 result.errors.append(_format_unsupported_soc_error(result.user_soc)) 

566 else: 

567 result.warnings.append( 

568 "No custom packages found. Checked ${ASCEND_HOME_PATH}/opp/vendors " 

569 "and ASCEND_CUSTOM_OPP_PATH." 

570 ) 

571 return result 

572 

573 

574def _format_unsupported_soc_error(user_soc: str) -> str: 

575 return "Error: SoC version is not supported: {}".format(user_soc) 

576 

577 

578def detect_conflicts( 

579 builtin_sources: Sequence[OpTypeSource], custom_sources: Sequence[OpTypeSource] 

580) -> ConflictReport: 

581 report = ConflictReport() 

582 builtin_all = set() 

583 for source in builtin_sources: 

584 builtin_all.update(source.optypes) 

585 builtin_reference = builtin_sources[0] if builtin_sources else None 

586 if builtin_reference: 

587 for custom in custom_sources: 

588 conflicts = sorted(custom.optypes.intersection(builtin_all)) 

589 if conflicts: 

590 report.custom_builtin.append( 

591 ConflictGroup( 

592 "Custom package conflicts with built-in OpTypes", 

593 custom, 

594 builtin_reference, 

595 conflicts, 

596 ) 

597 ) 

598 

599 for index, left in enumerate(custom_sources): 

600 for right in custom_sources[index + 1 :]: 

601 if left.root_path == right.root_path: 

602 continue 

603 conflicts = sorted(left.optypes.intersection(right.optypes)) 

604 if conflicts: 

605 report.custom_custom.append( 

606 ConflictGroup( 

607 "Custom package conflicts with another custom package", 

608 left, 

609 right, 

610 conflicts, 

611 ) 

612 ) 

613 return report 

614 

615 

616def _print_kv(label: str, value: Any, width: int = 22) -> None: 

617 print(" {label:<{width}} : {value}".format(label=label, width=width, value=value)) 

618 

619 

620def _format_table(headers: Sequence[str], rows: Sequence[Sequence[Any]]) -> List[str]: 

621 str_rows = [[str(item) for item in row] for row in rows] 

622 widths = [len(header) for header in headers] 

623 for row in str_rows: 

624 for index, item in enumerate(row): 

625 widths[index] = max(widths[index], len(item)) 

626 header_line = " " + " ".join( 

627 header.ljust(widths[index]) for index, header in enumerate(headers) 

628 ) 

629 separator = " " + " ".join("-" * widths[index] for index in range(len(headers))) 

630 body = [ 

631 " " + " ".join(item.ljust(widths[index]) for index, item in enumerate(row)) 

632 for row in str_rows 

633 ] 

634 return [header_line, separator] + body 

635 

636 

637def _print_scan_info(result: ScanResult) -> None: 

638 print("[Scan Info]") 

639 _print_kv("SoC", result.user_soc) 

640 _print_kv("ASCEND_HOME_PATH", result.ascend_home_path or "<unset>") 

641 _print_kv("ASCEND_CUSTOM_OPP_PATH", result.custom_opp_path or "<unset>") 

642 print("") 

643 

644 

645def _all_sources(result: ScanResult) -> List[OpTypeSource]: 

646 return result.builtin_sources + result.custom_sources 

647 

648 

649def _print_sources(result: ScanResult) -> None: 

650 print("[Sources]") 

651 rows = [ 

652 [ 

653 source.source_type, 

654 source.status, 

655 len(source.optypes), 

656 len(source.config_files), 

657 source.soc, 

658 source.root_path, 

659 ] 

660 for source in _all_sources(result) 

661 ] 

662 if rows: 

663 for line in _format_table( 

664 ["Type", "Status", "OpTypes", "ConfigFiles", "SoC", "Path"], rows 

665 ): 

666 print(line) 

667 else: 

668 print(" <no matching sources>") 

669 print("") 

670 

671 

672def _print_messages(title: str, messages: Iterable[str]) -> None: 

673 messages = [msg for msg in messages if msg] 

674 if not messages: 

675 return 

676 print("[{}]".format(title)) 

677 for msg in messages: 

678 for line in str(msg).splitlines(): 

679 print(" {}".format(line)) 

680 print("") 

681 

682 

683def _print_source_messages(result: ScanResult) -> None: 

684 warnings = list(result.warnings) 

685 errors = list(result.errors) 

686 for source in _all_sources(result): 

687 warnings.extend(source.warnings) 

688 errors.extend(source.errors) 

689 _print_messages("Warnings", warnings) 

690 _print_messages("Errors", errors) 

691 

692 

693def _print_list_summary(result: ScanResult, mode: str, count: int) -> None: 

694 print("[OpType List]") 

695 _print_kv("SoC", result.user_soc) 

696 _print_kv("Mode", mode) 

697 _print_kv("Total OpTypes", count) 

698 print("") 

699 

700 

701def _print_source_optypes(source: OpTypeSource) -> None: 

702 source_optypes = sorted(source.optypes) 

703 print(" [{type}] {path}".format(type=source.source_type, path=source.root_path)) 

704 _print_kv("OpTypes", len(source_optypes), width=16) 

705 _print_kv("SoC", source.soc, width=16) 

706 if source_optypes: 

707 print(" OpType") 

708 print(" ------") 

709 for optype in source_optypes: 

710 print(" {}".format(optype)) 

711 else: 

712 print(" <empty>") 

713 print("") 

714 

715 

716def _print_list(result: ScanResult, mode: str) -> None: 

717 if mode == BUILTIN: 

718 selected_sources = result.builtin_sources 

719 elif mode == CUSTOM: 

720 selected_sources = result.custom_sources 

721 else: 

722 selected_sources = _all_sources(result) 

723 optypes = sorted( 

724 {optype for source in selected_sources for optype in source.optypes} 

725 ) 

726 _print_list_summary(result, mode, len(optypes)) 

727 if len(selected_sources) > 1: 

728 for source in selected_sources: 

729 _print_source_optypes(source) 

730 else: 

731 if optypes: 

732 for line in _format_table(["OpType"], [[optype] for optype in optypes]): 

733 print(line) 

734 else: 

735 print(" <empty>") 

736 print("") 

737 

738 

739def _print_conflicts(result: ScanResult, report: ConflictReport) -> None: 

740 builtin_count = len( 

741 {optype for source in result.builtin_sources for optype in source.optypes} 

742 ) 

743 print("[Conflict Summary]") 

744 _print_kv("SoC", result.user_soc) 

745 _print_kv("Built-in OpTypes", builtin_count) 

746 _print_kv("Custom packages", len(result.custom_sources)) 

747 _print_kv("Custom vs Built-in", "{} group(s)".format(len(report.custom_builtin))) 

748 _print_kv("Custom vs Custom", "{} group(s)".format(len(report.custom_custom))) 

749 if not report.has_conflicts: 

750 _print_kv("Result", "No duplicate OpType conflicts found.") 

751 print("") 

752 

753 conflict_index = 1 

754 for group in report.custom_builtin + report.custom_custom: 

755 print("[Conflict {}] {}".format(conflict_index, group.conflict_type)) 

756 rows = [ 

757 [ 

758 "A", 

759 group.left.source_type, 

760 group.left.vendor_name or "-", 

761 group.left.root_path, 

762 ], 

763 [ 

764 "B", 

765 group.right.source_type, 

766 group.right.vendor_name or "-", 

767 group.right.root_path, 

768 ], 

769 ] 

770 for line in _format_table(["Pkg", "Type", "Vendor", "Path"], rows): 

771 print(line) 

772 _print_kv("Conflict count", len(group.optypes)) 

773 print(" Conflict OpTypes:") 

774 for optype in group.optypes: 

775 print(" - {}".format(optype)) 

776 print("") 

777 conflict_index += 1 

778 

779 

780def _build_parser() -> argparse.ArgumentParser: 

781 parser = argparse.ArgumentParser( 

782 prog="optype_collector",