Coverage for tsfpga/system_utils.py: 96%

71 statements  

« prev     ^ index     » next       coverage.py v7.3.2, created at 2023-12-10 20:51 +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 exists. 

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, this function will do nothing, since the directory already exists. 

122 

123 Return: 

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

125 """ 

126 if empty: 

127 delete(directory) 

128 elif directory.exists(): 

129 return directory 

130 

131 directory.mkdir(parents=True) 

132 return directory 

133 

134 

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

136 """ 

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

138 

139 Arguments: 

140 file_path: The file to be checked. 

141 directories: The directories to be controlled. 

142 

143 Return: 

144 True if there is a common path. 

145 """ 

146 for directory in directories: 

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

148 return True 

149 

150 return False 

151 

152 

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

154 """ 

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

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

157 

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

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

160 """ 

161 assert path.exists(), path 

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

163 

164 

165def run_command( 

166 cmd: list[str], 

167 cwd: Optional[Path] = None, 

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

169 capture_output: bool = False, 

170) -> subprocess.CompletedProcess[str]: 

171 """ 

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

173 

174 Arguments: 

175 cmd: The command to run. 

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

177 env: Environment variables to set. 

178 capture_output: Enable capturing or STDOUT and STDERR. 

179 

180 Return: 

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

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

183 be inspected. 

184 """ 

185 if not isinstance(cmd, list): 

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

187 

188 return subprocess.run( 

189 args=cmd, 

190 cwd=cwd, 

191 env=env, 

192 check=True, 

193 encoding=DEFAULT_FILE_ENCODING, 

194 capture_output=capture_output, 

195 ) 

196 

197 

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

199 """ 

200 Load the specified Python module. 

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

202 

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

204 """ 

205 python_module_name = file.stem 

206 

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

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

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

210 

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

212 spec.loader.exec_module(module=module) 

213 

214 return module 

215 

216 

217def system_is_windows() -> bool: 

218 """ 

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

220 """ 

221 return system() == "Windows"