()
| 26 | |
| 27 | |
| 28 | def npu_generate_uuid() -> str: |
| 29 | str_pid = str(os.getpid()) |
| 30 | npu_num = 8 |
| 31 | try: |
| 32 | for npu_id in range(npu_num): |
| 33 | cmd = ["npu-smi", "info", "-t", "proc-mem", "-i", str(npu_id)] |
| 34 | result = subprocess.run(cmd, check=True, capture_output=True, text=True) # noqa: S603 |
| 35 | str_result = str(result.stdout) |
| 36 | if str_pid in str_result: |
| 37 | # In A3 server, one NPU has two chips. |
| 38 | match_chip_count = re.search(r"Chip Count[^\d]*(\d+)", str_result) |
| 39 | chip_count = int(match_chip_count.group(1)) |
| 40 | search_after_pid = str_result[str_result.find(str_pid) + len(str_pid) :] |
| 41 | match_chip_id = re.search(r"Chip ID[^\d]*(\d+)", search_after_pid) |
| 42 | chip_id = int(match_chip_id.group(1)) |
| 43 | return f"{get_ip()}-{npu_id * chip_count + chip_id}" |
| 44 | raise ValueError("The current process is not running on the npu device") |
| 45 | except subprocess.CalledProcessError as e: |
| 46 | raise ValueError("The current process is not running on the npu device") from e |
| 47 | |
| 48 | |
| 49 | def _ibv_get_device_list() -> list[str]: |
no test coverage detected