Coverage for tsfpga/system_utils.py: 96%

72 statements  

« prev     ^ index     » next       coverage.py v7.5.1, created at 2024-05-07 11:31 +0000

1# -------------------------------------------------------------------------------------------------- 

2# Copyright (c) Lukas Vik. All rights reserved. 

3# 

4# This file is part of the tsfpga project, a project platform for modern FPGA development. 

5# https://tsfpga.com 

6# https://github.com/tsfpga/tsfpga 

7# -------------------------------------------------------------------------------------------------- 

8 

9# Standard libraries 

10import importlib.util 

11import os 

12import subprocess 

13from os.path import commonpath, relpath 

14from pathlib import Path 

15from platform import system 

16from shutil import rmtree 

17from typing import Any, Optional 

18 

19# First party libraries 

20from tsfpga import DEFAULT_FILE_ENCODING 

21 

22 

23def create_file(file: Path, contents: Optional[str] = None) -> Path: 

24 """ 

25 Create the ``file`` and any parent directories that do not exist. 

26 File will be empty unless ``contents`` is specified. 

27 

28 Return: 

29 The path to the file that was created (i.e. the original ``file`` argument). 

30 """ 

31 # Create directory unless it already exists. Do not delete anything if it does exist. 

32 create_directory(directory=file.parent, empty=False) 

33 

34 contents = "" if contents is None else contents 

35 with open(file, "w", encoding=DEFAULT_FILE_ENCODING) as file_handle: 

36 file_handle.write(contents) 

37 

38 return file 

39 

40 

41def read_file(file: Path) -> str: 

42 """ 

43 Read and return the file contents. 

44 """ 

45 with open(file, encoding=DEFAULT_FILE_ENCODING) as file_handle: 

46 return file_handle.read() 

47 

48 

49def read_last_lines_of_file(file: Path, num_lines: int) -> str: 

50 """ 

51 Read a number of lines from the end of a file, without buffering the whole file. 

52 Similar to unix ``tail`` command. 

53 

54 Arguments: 

55 file: The file that shall be read. 

56 num_lines: The number of lines to read. 

57 

58 Return: 

59 The last lines of the file. 

60 """ 

61 result_lines: list[str] = [] 

62 blocks_to_read = 0 

63 

64 with open(file, encoding=DEFAULT_FILE_ENCODING) as file_handle: 

65 while len(result_lines) < num_lines: 

66 # Since we do not know the line lengths, there is some guessing involved. Keep reading 

67 # larger and larger blocks until we have all the lines that are requested. 

68 blocks_to_read += 1 

69 

70 try: 

71 # Read a block from the end 

72 file_handle.seek(-blocks_to_read * 4096, os.SEEK_END) 

73 except IOError: 

74 # Tried to read more data than what is available. Read whatever we have and return 

75 # to user. 

76 file_handle.seek(0) 

77 result_lines = file_handle.readlines() 

78 break 

79 

80 result_lines = file_handle.readlines() 

81 

82 result = "".join(result_lines[-num_lines:]) 

83 return result 

84 

85 

86def delete(path: Path, wait_until_deleted: bool = False) -> Path: 

87 """ 

88 Delete a file or directory from the filesystem. 

89 

90 Arguments: 

91 path: The file/directory to be deleted. 

92 wait_until_deleted: When set to ``True``, the function will poll the filesystem 

93 after initiating the deletion, and not return until the path is in fact deleted. 

94 Is needed on some filesystems/mounts in a situation where we delete a path and 

95 then directly want to write to it afterwards. 

96 

97 Return: 

98 The path that was deleted (i.e. the original ``path`` argument). 

99 """ 

100 if path.exists(): 

101 if path.is_dir(): 

102 rmtree(path) 

103 else: 

104 path.unlink() 

105 

106 if wait_until_deleted: 

107 while path.exists(): 

108 pass 

109 

110 return path 

111 

112 

113def create_directory(directory: Path, empty: bool = True) -> Path: 

114 """ 

115 Create a directory. 

116 

117 Arguments: 

118 directory: Path to the directory. 

119 empty: If true and the directory already exists, all existing files/folders in it will 

120 be deleted. 

121 If false, the directory will be left as-is if it already exists, or will be newly 

122 created if it does not. 

123 

124 Return: 

125 The path that was created (i.e. the original ``directory`` argument). 

126 """ 

127 if empty: 

128 # Delete directory, and anything inside it, before creating it below. 

129 delete(directory) 

130 

131 elif directory.exists(): 

132 if directory.is_dir(): 

133 return directory 

134 

135 raise FileExistsError(f"Requested directory path already exists as a file: {directory}") 

136 

137 directory.mkdir(parents=True) 

138 return directory 

139 

140 

141def file_is_in_directory(file_path: Path, directories: list[Path]) -> bool: 

142 """ 

143 Check if the file is in any of the directories. 

144 

145 Arguments: 

146 file_path: The file to be checked. 

147 directories: The directories to be controlled. 

148 

149 Return: 

150 True if there is a common path. 

151 """ 

152 for directory in directories: 

153 if commonpath([str(file_path), str(directory)]) == str(directory): 

154 return True 

155 

156 return False 

157 

158 

159def path_relative_to(path: Path, other: Path) -> Path: 

160 """ 

161 Return a relative path from ``other`` to ``path``. 

162 This function works even if ``path`` is not inside a ``other`` folder. 

163 

164 Note Path.relative_to() does not support the use case where e.g. readme.md should get 

165 relative path "../readme.md". Hence we have to use os.path. 

166 """ 

167 return Path(relpath(str(path), str(other))) 

168 

169 

170def run_command( 

171 cmd: list[str], 

172 cwd: Optional[Path] = None, 

173 env: Optional[dict[str, str]] = None, 

174 capture_output: bool = False, 

175) -> subprocess.CompletedProcess[str]: 

176 """ 

177 Will raise ``CalledProcessError`` if the command fails. 

178 

179 Arguments: 

180 cmd: The command to run. 

181 cwd: The working directory where the command shall be executed. 

182 env: Environment variables to set. 

183 capture_output: Enable capturing of STDOUT and STDERR. 

184 

185 Return: 

186 Returns the subprocess completed process object, which contains useful information. 

187 If ``capture_output`` is set, the ``stdout`` and ``stderr`` members of this object can 

188 be inspected. 

189 """ 

190 if not isinstance(cmd, list): 

191 raise ValueError("Must be called with a list, not a string") 

192 

193 return subprocess.run( 

194 args=cmd, 

195 cwd=cwd, 

196 env=env, 

197 check=True, 

198 encoding=DEFAULT_FILE_ENCODING, 

199 capture_output=capture_output, 

200 ) 

201 

202 

203def load_python_module(file: Path) -> Any: 

204 """ 

205 Load the specified Python module. 

206 Note that in Python nomenclature, a module is a source code file. 

207 

208 On the returned object, you can call functions or instantiate classes that are in the module. 

209 """ 

210 python_module_name = file.stem 

211 

212 spec = importlib.util.spec_from_file_location(name=python_module_name, location=file) 

213 if spec is None or spec.loader is None: 

214 raise RuntimeError(f"Could not load the Python module: {file}") 

215 

216 module = importlib.util.module_from_spec(spec=spec) 

217 spec.loader.exec_module(module=module) 

218 

219 return module 

220 

221 

222def system_is_windows() -> bool: 

223 """ 

224 Return True if the script is being executed on a computer running the Windows operating system. 

225 """ 

226 return system() == "Windows"