1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
# SPDX-FileCopyrightText: Copyright 2022-2024, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Utils related to file management."""
from __future__ import annotations
import hashlib
import importlib.resources as pkg_resources
import os
import shutil
from contextlib import contextmanager
from pathlib import Path
from tempfile import mkstemp
from tempfile import TemporaryDirectory
from typing import Generator
from typing import Iterable
USER_ONLY_PERM_MASK = 0o700
def get_mlia_resources() -> Path:
"""Get the path to the resources directory."""
with pkg_resources.path("mlia", "__init__.py") as init_path:
project_root = init_path.parent
return project_root / "resources"
def get_vela_config() -> Path:
"""Get the path to the default Vela config file."""
return get_mlia_resources() / "vela/vela.ini"
def get_mlia_target_profiles_dir() -> Path:
"""Get the profiles file."""
return get_mlia_resources() / "target_profiles"
@contextmanager
def temp_file(suffix: str | None = None) -> Generator[Path, None, None]:
"""Create temp file and remove it after."""
_, tmp_file = mkstemp(suffix=suffix)
try:
yield Path(tmp_file)
finally:
os.remove(tmp_file)
@contextmanager
def temp_directory(suffix: str | None = None) -> Generator[Path, None, None]:
"""Create temp directory and remove it after."""
with TemporaryDirectory(suffix=suffix) as tmpdir:
yield Path(tmpdir)
def file_chunks(
filepath: str | Path, chunk_size: int = 4096
) -> Generator[bytes, None, None]:
"""Return sequence of the file chunks."""
with open(filepath, "rb") as file:
while data := file.read(chunk_size):
yield data
def hexdigest(
filepath: str | Path, hash_obj: hashlib._Hash # pylint: disable=no-member
) -> str:
"""Return hex digest of the file."""
for chunk in file_chunks(filepath):
hash_obj.update(chunk)
return hash_obj.hexdigest()
def sha256(filepath: Path) -> str:
"""Return SHA256 hash of the file."""
return hexdigest(filepath, hashlib.sha256())
def all_files_exist(paths: Iterable[Path]) -> bool:
"""Check if all files exist."""
return all(item.is_file() for item in paths)
def all_paths_valid(paths: Iterable[Path]) -> bool:
"""Check if all paths are valid."""
return all(item.exists() for item in paths)
def copy_all(*paths: Path, dest: Path) -> None:
"""Copy files/directories into destination folder."""
dest.mkdir(exist_ok=True)
for path in paths:
if path.is_file():
shutil.copy2(path, dest)
if path.is_dir():
shutil.copytree(path, dest, dirs_exist_ok=True)
def recreate_directory(dir_path: Path, mode: int = USER_ONLY_PERM_MASK) -> None:
"""Recreate directory."""
if dir_path.exists():
if not dir_path.is_dir():
raise ValueError(f"Path {dir_path} is not a directory.")
shutil.rmtree(dir_path)
dir_path.mkdir(exist_ok=True, mode=mode)
@contextmanager
def working_directory(
working_dir: Path, create_dir: bool = False
) -> Generator[Path, None, None]:
"""Temporary change working directory."""
current_working_dir = Path.cwd()
if create_dir:
working_dir.mkdir()
os.chdir(working_dir)
try:
yield working_dir
finally:
os.chdir(current_working_dir)
|
