Coverage for tsfpga/system_utils.py: 95%

76 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-02-21 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 

9from __future__ import annotations 

10 

11import importlib.util 

12import os 

13import subprocess 

14from os.path import commonpath, relpath 

15from pathlib import Path 

16from platform import system 

17from shutil import rmtree 

18from sys import modules 

19from typing import TYPE_CHECKING 

20 

21from tsfpga import DEFAULT_FILE_ENCODING 

22 

23if TYPE_CHECKING: 

24 from types import ModuleType 

25 

26 

27def create_file(file: Path, contents: str | None = None) -> Path: 

28 """ 

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

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

31 

32 Return: 

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

34 """ 

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

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

37 

38 contents = "" if contents is None else contents 

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

40 file_handle.write(contents) 

41 

42 return file 

43 

44 

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

46 """ 

47 Read and return the file contents. 

48 """ 

49 with file.open(encoding=DEFAULT_FILE_ENCODING) as file_handle: 

50 return file_handle.read() 

51 

52 

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

54 """ 

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

56 Similar to unix ``tail`` command. 

57 

58 Arguments: 

59 file: The file that shall be read. 

60 num_lines: The number of lines to read. 

61 

62 Return: 

63 The last lines of the file. 

64 """ 

65 result_lines: list[str] = [] 

66 blocks_to_read = 0 

67 

68 with file.open(encoding=DEFAULT_FILE_ENCODING) as file_handle: 

69 while len(result_lines) < num_lines: 

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

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

72 blocks_to_read += 1 

73 

74 try: 

75 # Read a block from the end 

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

77 except OSError: 

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

79 # to user. 

80 file_handle.seek(0) 

81 result_lines = file_handle.readlines() 

82 break 

83 

84 result_lines = file_handle.readlines() 

85 

86 return "".join(result_lines[-num_lines:]) 

87 

88 

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

90 """ 

91 Delete a file or directory from the filesystem. 

92 

93 Arguments: 

94 path: The file/directory to be deleted. 

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

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

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

98 then directly want to write to it afterwards. 

99 

100 Return: 

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

102 """ 

103 if path.exists(): 

104 if path.is_dir(): 

105 rmtree(path) 

106 else: 

107 path.unlink() 

108 

109 if wait_until_deleted: 

110 while path.exists(): 

111 pass 

112 

113 return path 

114 

115 

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

117 """ 

118 Create a directory. 

119 

120 Arguments: 

121 directory: Path to the directory. 

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

123 be deleted. 

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

125 created if it does not. 

126 

127 Return: 

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

129 """ 

130 if empty: 

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

132 delete(directory) 

133 

134 elif directory.exists(): 

135 if directory.is_dir(): 

136 return directory 

137 

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

139 

140 directory.mkdir(parents=True) 

141 return directory 

142 

143 

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

145 """ 

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

147 

148 Arguments: 

149 file_path: The file to be checked. 

150 directories: The directories to be controlled. 

151 

152 Return: 

153 True if there is a common path. 

154 """ 

155 for directory in directories: 

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

157 return True 

158 

159 return False 

160 

161 

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

163 """ 

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

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

166 

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

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

169 """ 

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

171 

172 

173def run_command( 

174 cmd: list[str], 

175 cwd: Path | None = None, 

176 env: dict[str, str] | None = None, 

177 capture_output: bool = False, 

178) -> subprocess.CompletedProcess[str]: 

179 """ 

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

181 

182 Arguments: 

183 cmd: The command to run. 

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

185 env: Environment variables to set. 

186 capture_output: Enable capturing of STDOUT and STDERR. 

187 

188 Return: 

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

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

191 be inspected. 

192 """ 

193 if not isinstance(cmd, list): 

194 raise TypeError("Must be called with a list, not a string") 

195 

196 return subprocess.run( 

197 args=cmd, 

198 cwd=cwd, 

199 env=env, 

200 check=True, 

201 encoding=DEFAULT_FILE_ENCODING, 

202 capture_output=capture_output, 

203 ) 

204 

205 

206def load_python_module(file: Path) -> ModuleType: 

207 """ 

208 Load the specified Python module. 

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

210 

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

212 """ 

213 python_module_name = file.stem 

214 

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

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

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

218 

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

220 modules[python_module_name] = module 

221 spec.loader.exec_module(module=module) 

222 

223 return module 

224 

225 

226def system_is_windows() -> bool: 

227 """ 

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

229 """ 

230 return system() == "Windows"