skills/get-available-resources/scripts/detect_resources.py

• _warning · function · L98-L113 — def _warning( warnings: list[dict[str, str]], code: str, component: str, message: str, *, severity: str = "warning", ) -> None
• _provenance · function · L116-L124 — def _provenance( records: list[dict[str, str]], component: str, source: str, status: str, ) -> None
• _safe_text · function · L127-L132 — def _safe_text(value: Any, maximum: int = 128) -> str | None
• _read_bounded_text · function · L135-L140 — def _read_bounded_text(path: Path, maximum: int = MAX_KERNEL_TEXT) -> str
• _try_read · function · L143-L150 — def _try_read( read_text: Callable[[Path], str], path: Path, ) -> str | None
• _bounded_nonnegative · function · L153-L160 — def _bounded_nonnegative(value: str | int | None) -> int | None
• parse_cpu_list · function · L163-L188 — def parse_cpu_list(value: str) -> int | None
• _parse_cpu_max · function · L191-L202 — def _parse_cpu_max(value: str) -> float | None
• _run_bounded_command · function · L205-L291 — def _run_bounded_command( argv: Sequence[str], *, timeout: float = 4.0, maximum: int = MAX_PROBE_OUTPUT, ) -> dict[str, Any]
• drain · function · L232-L251 — def drain(name: str, stream: Any) -> None
• _load_psutil · function · L294-L311 — def _load_psutil( warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> Any | None
• _linux_physical_cores · function · L314-L330 — def _linux_physical_cores( read_text: Callable[[Path], str], ) -> int | None
• _mac_sysctl_values · function · L333-L365 — def _mac_sysctl_values( run_command: Callable[..., dict[str, Any]], warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, Any]
• _detect_cpu_inventory · function · L368-L427 — def _detect_cpu_inventory( *, system: str, psutil_module: Any | None, mac_sysctl: Mapping[str, Any], read_text: Callable[[Path], str], warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, int | None]
• _detect_process_cpu_count · function · L430-L492 — def _detect_process_cpu_count( *, psutil_module: Any | None, warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, int | None]
• _cgroup_relative_path · function · L495-L504 — def _cgroup_relative_path(payload: str) -> tuple[str, ...] | None
• detect_cgroup_v2 · function · L507-L626 — def detect_cgroup_v2( *, read_text: Callable[[Path], str] = _read_bounded_text, root: Path = Path("/sys/fs/cgroup"), proc_self_cgroup: Path = Path("/proc/self/cgroup"), warnings: list[dict[str, str]] | None = None, provenance: list[dict[str, str]] | None = None, ) -> dict[str, Any]
• _parse_slurm_count · function · L629-L640 — def _parse_slurm_count(value: str | None) -> int | None
• _parse_slurm_memory · function · L643-L655 — def _parse_slurm_memory(value: str | None) -> int | None
• _parse_first_repeated_count · function · L658-L665 — def _parse_first_repeated_count(value: str | None) -> int | None
• detect_scheduler · function · L668-L772 — def detect_scheduler( environ: Mapping[str, str], *, warnings: list[dict[str, str]] | None = None, provenance: list[dict[str, str]] | None = None, ) -> dict[str, Any]
• _effective_cpu · function · L775-L809 — def _effective_cpu( host: Mapping[str, int | None], process: Mapping[str, int | None], cgroup: Mapping[str, Any], scheduler: Mapping[str, Any], ) -> dict[str, Any]
• _parse_linux_meminfo · function · L812-L823 — def _parse_linux_meminfo(payload: str) -> dict[str, int | None]
• _detect_memory · function · L826-L979 — def _detect_memory( *, system: str, machine: str, psutil_module: Any | None, mac_sysctl: Mapping[str, Any], cgroup: Mapping[str, Any], scheduler: Mapping[str, Any], read_text: Callable[[Path], str], warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, Any]
• _detect_disk · function · L982-L1039 — def _detect_disk( warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, Any]
• _mib_to_bytes · function · L1042-L1050 — def _mib_to_bytes(value: str) -> int | None
• parse_nvidia_csv · function · L1053-L1086 — def parse_nvidia_csv(payload: str) -> list[dict[str, Any]]
• _recursive_scalar · function · L1089-L1107 — def _recursive_scalar( value: Any, accepted_keys: set[str], ) -> tuple[str, Any] | None
• _amd_entries · function · L1110-L1149 — def _amd_entries(data: Any) -> list[tuple[int, dict[str, Any]]]
• _amd_memory_bytes · function · L1152-L1178 — def _amd_memory_bytes(entry: Mapping[str, Any]) -> int | None
• parse_amd_json · function · L1181-L1221 — def parse_amd_json(payload: str) -> list[dict[str, Any]]
• parse_apple_profiler_json · function · L1224-L1267 — def parse_apple_profiler_json( payload: str, *, machine: str, ) -> list[dict[str, Any]]
• _visibility_summary · function · L1270-L1308 — def _visibility_summary(environ: Mapping[str, str]) -> dict[str, Any]
• _accelerator_upper_bounds · function · L1311-L1345 — def _accelerator_upper_bounds( devices: Sequence[Mapping[str, Any]], visibility: Mapping[str, Mapping[str, Any]], scheduler: Mapping[str, Any], ) -> dict[str, int | None]
• _record_probe_failure · function · L1348-L1366 — def _record_probe_failure( result: Mapping[str, Any], *, code: str, component: str, tool: str, warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> None
• _detect_accelerators · function · L1369-L1519 — def _detect_accelerators( *, system: str, machine: str, environ: Mapping[str, str], scheduler: Mapping[str, Any], run_command: Callable[..., dict[str, Any]], warnings: list[dict[str, str]], provenance: list[dict[str, str]], ) -> dict[str, Any]
• _container_context · function · L1522-L1554 — def _container_context( *, cgroup: Mapping[str, Any], exists: Callable[[Path], bool] = Path.exists, ) -> dict[str, Any]
• _observed_at · function · L1557-L1563 — def _observed_at() -> str
• collect_snapshot · function · L1566-L1716 — def collect_snapshot( *, skip_accelerators: bool = False, observed_at: str | None = None, environ: Mapping[str, str] | None = None, psutil_module: Any = _AUTO, read_text: Callable[[Path], str] = _read_bounded_text, run_command: Callable[..., dict[str, Any]] = _run_bounded_command, ) -> dict[str, Any]
• detect_all_resources · function · L1719-L1724 — def detect_all_resources(output_path: str | None = None) -> dict[str, Any]
• build_parser · function · L1727-L1750 — def build_parser() -> argparse.ArgumentParser
• main · function · L1753-L1763 — def main(argv: Sequence[str] | None = None) -> int