diff --git a/README.md b/README.md index 9bd0296..7490a22 100644 --- a/README.md +++ b/README.md @@ -10,8 +10,6 @@ The advantages of this tool are : * hash system to avoid time check issues (same file content, remote files problems, etc) -* watch option to recompile whenever a source is changed - ## Use @@ -27,13 +25,7 @@ This repository is meant to be used as a submodule of a C/C++ project. By default 3 targets can be used `make` (or `make all` as usual default of makefiles), `make debug` and `make clean`. -**Example :** from the root folder of a project : - -``` -git submodule add https://gitlab.com/corentin-pro/umake.git && cp umake/make* . -``` - -Then change the configuration in `make.py` and a simple `make` command will build your project! +**Be careful** : `clean` will delete the object and binary output directories defined in `make.py`. ## How it works @@ -43,13 +35,19 @@ Then change the configuration in `make.py` and a simple `make` command will buil µmake expects all the header and source files to be place in a directories specified in the configuration (in `make.py`). Object files and final binaries are outputed also in directories specified in the configuration but objects files in `debug` and `release` (default) are separated. -The separation of object/release objects files is meant to avoid rebuilding the whole project after debugging a single file. +The separation of objects files is meant to switch mode quickly (only binaries are rebuilt). To determine if a source file needs to be recompiled an hash is saved. The hashed are saved in a JSON file at the root of the object files directory (there are in fact 2 seperate files respectively in `debug` oand `release`). +If any header file changes every source files will be recompiled to avoid any issue (hence the small or medium-size project target for this tool). This could be address in the future if needed. + ### Caveats +* Header files dependencies is not implemented (any change triggers all compilation) + * Knowing python is recommended to understand the tool for any usage above configuration change. +* Targets management is not yet implemented (building all binaries), this can be easily address knowing python. A generic solution should be possible in the future. + * Template configuration is meant for C++ files (cpp/hpp) for now. diff --git a/make.py b/make.py index fb88c37..72c1ded 100644 --- a/make.py +++ b/make.py @@ -8,12 +8,11 @@ from umake import make class Config: CC = 'g++' # Compiler to call - APPS = ['app_name'] # Output binaries (path to source without extension) + APPS = ['app_name'] # Output binaries (need to be found as .cpp directly in SOURCE_DIR) IGNORE_APPS = [] JOB_COUNT = int(os.cpu_count() * 0.8) # Concurent jobs (multi-processing) - WATCH = False # Watch source modification for auto-compiling - BIN_DIR = Path('bin') # Output directory (binaries) + BIN_DIR = Path('bin') # Output directory (binaries) INCLUDE_DIR = Path('include') # Include directory (header files) OBJECT_DIR = Path('obj') # Temporary directory (object files) SOURCE_DIR = Path('src') # Source directories @@ -24,9 +23,8 @@ class Config: COMPILE_FLAGS = f'-Wall -I{INCLUDE_DIR}' # Flags added for compiling (recommandation : `pkg-config --cflags`) LINK_FLAGS = '' # Flags added for linking (recommandation : `pkg-config --libs`) - PRE_COMPILE_FUNCTION = None # Function to run before compile (not call in --clean situation) - - CPP_SOURCES = [filepath for filepath in SOURCE_DIR.rglob('*.cpp') if not filepath.name.startswith('.')] + CPP_HEADERS = INCLUDE_DIR.rglob('*.hpp') + CPP_SOURCES = SOURCE_DIR.rglob('*.cpp') def main(): diff --git a/makefile b/makefile index 4d5ae1c..db2db6a 100644 --- a/makefile +++ b/makefile @@ -1,13 +1,13 @@ MAKE_PATH="." UMAKE_PATH="umake" -.PHONY: all debug clean +.PHONY: all clean all: - @PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py + @PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py debug: - @PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py --type debug + @PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py --type debug clean: - @PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py --clean + @PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py --clean diff --git a/setup.cfg b/setup.cfg deleted file mode 100644 index c6cb7d1..0000000 --- a/setup.cfg +++ /dev/null @@ -1,25 +0,0 @@ -[flake8] -max-line-length=120 -ignore=D10,D203,D204 - -[pycodestyle] -max-line-length=120 -ignore=D10,D203,D204 - -[pylint.DESIGN] -max-args=16 -min-public-methods=0 -max-attributes=16 -max-locals=64 - -[pylint.FORMAT] -max-line-length=120 - -[pylint.MESSAGE CONTROL] -disable=missing-module-docstring, missing-function-docstring, missing-class-docstring, relative-beyond-top-level, too-few-public-methods, import-error - -[pylint.SIMILARITIES] -min-similarity-lines=6 - -[pydocstyle] -ignore=D10,D203,D204 diff --git a/umake.py b/umake.py index a2bf31f..56e6050 100755 --- a/umake.py +++ b/umake.py @@ -13,9 +13,7 @@ import sys class Config: CC = 'g++' # Compiler to call APPS = ['app_name'] # Output binaries (need to be found as .cpp directly in SOURCE_DIR) - IGNORE_APPS = [] JOB_COUNT = int(os.cpu_count() * 0.8) # Concurent jobs (multi-processing) - WATCH = False # Watch source modification for auto-compiling BIN_DIR = Path('bin') # Output directory (binaries) INCLUDE_DIR = Path('include') # Include directory (header files) @@ -28,14 +26,12 @@ class Config: COMPILE_FLAGS = f'-Wall -I{INCLUDE_DIR}' # Flags added for compiling (recommandation : `pkg-config --cflags`) LINK_FLAGS = '' # Flags added for linking (recommandation : `pkg-config --libs`) - PRE_COMPILE_FUNCTION = None # Function to run before compile (not call in --clean situation) - + CPP_HEADERS = INCLUDE_DIR.rglob('*.hpp') CPP_SOURCES = SOURCE_DIR.rglob('*.cpp') class ConsoleColor: - """Simple shortcut to use colors in console.""" - + """Simple shortcut to use colors in console""" HEADER = '\033[95m' BLUE = '\033[94m' GREEN = '\033[92m' @@ -47,8 +43,8 @@ class ConsoleColor: def get_hash(path: Path) -> str: - hash_obj = hashlib.md5() - with open(path, 'r', encoding='utf-8') as hashing_file: + with open(path, 'r') as hashing_file: + hash_obj = hashlib.md5() hash_obj.update(hashing_file.read().encode()) return hash_obj.hexdigest() @@ -62,19 +58,10 @@ def make(config: Config): # Clean action if arguments.clean: - if config.OBJECT_DIR.exists(): - for object_entry in config.OBJECT_DIR.iterdir(): - shutil.rmtree(object_entry, ignore_errors=True) - if config.BIN_DIR.exists(): - for binary_entry in config.BIN_DIR.iterdir(): - shutil.rmtree(binary_entry, ignore_errors=True) + shutil.rmtree(config.OBJECT_DIR, ignore_errors=True) + shutil.rmtree(config.BIN_DIR, ignore_errors=True) return - if config.PRE_COMPILE_FUNCTION is not None: - config.PRE_COMPILE_FUNCTION() - if 'IGNORE_APPS' in config.__dict__: - config.IGNORE_APPS = [] - # Update flags and directories for mode debug/release if arguments.type == 'debug': config.COMMON_FLAGS += ' ' + config.COMMON_DEBUG_FLAGS @@ -83,223 +70,158 @@ def make(config: Config): config.COMMON_FLAGS += ' ' + config.COMMON_RELEASE_FLAGS config.OBJECT_DIR = config.OBJECT_DIR / 'release' - # Update job count - config.JOB_COUNT = arguments.j + # Create list of source to process (tuple[source_path, object_path]) + compile_list = [(Path(source_file), + config.OBJECT_DIR / source_file.parent.relative_to(config.SOURCE_DIR) / (source_file.stem + '.o')) + for source_file in config.CPP_SOURCES] + todo_list: list[tuple[Path, str]] = [] + error_paths: list[tuple[Path, str]] = [] - Builder(config).make() + hash_dict = {} + if (config.OBJECT_DIR / 'hash.json').exists(): + with open(config.OBJECT_DIR / 'hash.json', 'r') as hash_file: + hash_dict = json.loads(hash_file.read()) + # Get source dependencies + dependency_dict = {} + for source_path, object_path in compile_list: + cmd = ' '.join([config.CC, config.COMMON_FLAGS, config.COMPILE_FLAGS, str(source_path), '-M']) + job = subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True, shell=True) + if job.returncode != 0: + error_paths.append((source_path, job.stdout.read() + job.stderr.read())) + break + dependency_dict[str(source_path)] = job.stdout.split('.o: ')[1].replace('\n', '').replace('\\ ', '').split(' ') + if error_paths: + for error_path, error_text in error_paths: + print(ConsoleColor.RED + f'Error checking dependencies for {error_path}:\n' + + ConsoleColor.ENDCOLOR + error_text) + with open(config.OBJECT_DIR / 'hash.json', 'w') as hash_file: + hash_file.write(json.dumps(hash_dict, indent=1)) + sys.exit(1) -class Builder: - def __init__(self, config: Config): - self._config = config + # Check source file and generate compilation commands to execute + for source_path, object_path in compile_list: + cmd = ' '.join([config.CC, config.COMMON_FLAGS, config.COMPILE_FLAGS, + str(source_path), '-c', '-o', str(object_path)]) + if not object_path.parent.exists(): + object_path.parent.mkdir(parents=True) + source_hash = get_hash(source_path) + dependency_changed = False + for dependency_path in dependency_dict[str(source_path)]: + dependency_hash = get_hash(dependency_path) + if str(dependency_path) not in hash_dict or hash_dict[str(dependency_path)] != dependency_hash: + dependency_changed = True + print(ConsoleColor.ORANGE + f'Dependency changed for {source_path} : {dependency_path}' + + ConsoleColor.ENDCOLOR) + break + if (dependency_changed or not object_path.exists() + or str(source_path) not in hash_dict or hash_dict[str(source_path)] != source_hash): + todo_list.append((source_path, cmd)) + continue - # Create list of source to process (tuple[source_path, object_path]) - self._compile_dict: dict[Path, Path] = { - Path(source_file): (self._config.OBJECT_DIR / source_file.parent.relative_to(self._config.SOURCE_DIR) - / (source_file.stem + '.o')) - for source_file in self._config.CPP_SOURCES} - self._todo_dict: dict[Path, str] = {} + if not todo_list: + print(ConsoleColor.GREEN + 'Nothing to do' + ConsoleColor.ENDCOLOR) + return - self._hash_dict: dict[str, str] = {} - if (self._config.OBJECT_DIR / 'hash.json').exists(): - with open(self._config.OBJECT_DIR / 'hash.json', 'r', encoding='utf-8') as hash_file: - self._hash_dict = json.loads(hash_file.read()) - - # Get source dependencies - error_paths: list[tuple[Path, str]] = [] - self._dependency_dict: dict[str, list[str]] = {} - for source_path, _object_path in self._compile_dict.items(): - cmd = ' '.join( - [self._config.CC, self._config.COMMON_FLAGS, self._config.COMPILE_FLAGS, str(source_path), '-M']) + # Running compilation processes + if not os.path.exists(config.OBJECT_DIR): + os.makedirs(config.OBJECT_DIR) + if arguments.j > 1: # Multi-process + jobs: list[tuple[Path, subprocess.Popen]] = [] + for source_path, cmd in todo_list: + print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) + jobs.insert(0, (source_path, subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, + shell=True))) # FIFO style (will be poped) + if len(jobs) >= arguments.j: # If jobs count is maxed we wait for the oldest one to finished + job_path, oldest_job = jobs.pop() + oldest_job.wait() + if oldest_job.returncode != 0: + error_paths.append((job_path, oldest_job.stdout.read() + oldest_job.stderr.read())) + break + # Update hash if no error + for dependency_path in dependency_dict[str(job_path)]: + hash_dict[str(dependency_path)] = get_hash(dependency_path) + for job_path, job in jobs: # Wait the last jobs to finish + job.wait() + if job.returncode != 0: + error_paths.append((job_path, job.stdout.read() + job.stderr.read())) + else: + # Update hash if no error + for dependency_path in dependency_dict[str(job_path)]: + hash_dict[str(dependency_path)] = get_hash(dependency_path) + else: # Single-process + for source_path, source_hash, cmd in todo_list: + print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) job = subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, shell=True) if job.returncode != 0: error_paths.append((source_path, job.stdout + job.stderr)) break - self._dependency_dict[str(source_path)] = [ - line for line in job.stdout.split('.o: ')[1].replace('\n', '').replace('\\ ', '').split(' ') if line] - if error_paths: - for error_path, error_text in error_paths: - print(f'{ConsoleColor.RED}Error checking dependencies for {error_path}:' - f'\n{ConsoleColor.ENDCOLOR + error_text}') - with open(self._config.OBJECT_DIR / 'hash.json', 'w', encoding='utf-8') as hash_file: - hash_file.write(json.dumps(self._hash_dict, indent=1)) - sys.exit(1) + # Update hash if no error + for dependency_path in dependency_dict[str(source_path)]: + hash_dict[str(dependency_path)] = get_hash(dependency_path) + if error_paths: + for error_path, error_text in error_paths: + print(ConsoleColor.RED + f'Error compiling {error_path}:\n' + ConsoleColor.ENDCOLOR + error_text) + with open(config.OBJECT_DIR / 'hash.json', 'w') as hash_file: + hash_file.write(json.dumps(hash_dict, indent=1)) + sys.exit(1) - def make(self): - self._populate_todo_dict() - - if self._config.WATCH: - from watch import Watcher, WatchFlag # pylint: disable=import-outside-toplevel - - def callback(path: Path, _flags: WatchFlag): - source_hash = get_hash(path) - if str(path) not in self._hash_dict or source_hash != self._hash_dict[str(path)]: - print(f'{ConsoleColor.ORANGE}Source file changed : {path} {ConsoleColor.ENDCOLOR}') - self._populate_todo_dict() - self._compile_sources() - - self._compile_sources() - watcher = Watcher() - for source_path in self._compile_dict: - watcher.register(source_path, WatchFlag.MODIFY, callback) - try: - watcher.watch() - except KeyboardInterrupt: - print('\rExit') - elif not self._compile_sources(): - sys.exit(1) - - def _populate_todo_dict(self): - """Check source file and generate compilation commands to execute.""" - for source_path, object_path in self._compile_dict.items(): - if not object_path.parent.exists(): - object_path.parent.mkdir(parents=True) - source_hash = get_hash(source_path) - dependency_changed = False - for path in self._dependency_dict[str(source_path)]: - dependency_hash = get_hash(path) - if str(path) not in self._hash_dict or self._hash_dict[str(path)] != dependency_hash: - dependency_changed = True - print(f'{ConsoleColor.ORANGE}Dependency changed for {source_path} : {path} {ConsoleColor.ENDCOLOR}') - break - if (dependency_changed or not object_path.exists() - or str(source_path) not in self._hash_dict or self._hash_dict[str(source_path)] != source_hash): - self._todo_dict[source_path] = self._generate_compilation_command(source_path, object_path) + # Running linking processes + if not config.BIN_DIR.exists(): + config.BIN_DIR.mkdir(parents=True) + all_app_objects = [config.OBJECT_DIR / Path(app_path).parent / (Path(app_path).stem + '.o') + for app_path in config.APPS] + if arguments.j > 1: # Multi-process + jobs: list[tuple[Path, subprocess.Popen]] = [] + for app_path, app_object_path in zip(config.APPS, all_app_objects): + if 'IGNORE_APPS' in config.__dict__ and app_path in config.IGNORE_APPS: continue - - def _generate_compilation_command(self, source_path: Path, object_path: Path) -> str: - return ' '.join([self._config.CC, self._config.COMMON_FLAGS, self._config.COMPILE_FLAGS, - str(source_path), '-c', '-o', str(object_path)]) - - def _compile_sources(self) -> bool: - if not self._todo_dict and all( - (self._config.BIN_DIR / app_path).exists() - for app_path in self._config.APPS if app_path not in self._config.IGNORE_APPS): - print(f'{ConsoleColor.GREEN}Nothing to do{ConsoleColor.ENDCOLOR}') - return True - - # Running compilation processes - if not os.path.exists(self._config.OBJECT_DIR): - os.makedirs(self._config.OBJECT_DIR) - error_paths: list[tuple[Path, str]] = [] - if self._config.JOB_COUNT > 1: # Multi-process - jobs: list[tuple[Path, subprocess.Popen]] = [] - completed_paths: list[Path] = [] - for source_path, cmd in self._todo_dict.items(): - print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) - jobs.insert(0, (source_path, subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, - shell=True))) # FIFO style (will be poped) - if len(jobs) >= self._config.JOB_COUNT: # If jobs count is maxed we wait for the oldest one to finished - job_path, oldest_job = jobs.pop() - oldest_job.wait() - if oldest_job.returncode != 0: - error_paths.append((job_path, oldest_job.stdout.read() + oldest_job.stderr.read())) - if str(job_path) in self._hash_dict: - del self._hash_dict[str(job_path)] - break - # Update hash if no error - for dependency_path in self._dependency_dict[str(job_path)]: - self._hash_dict[str(dependency_path)] = get_hash(dependency_path) - completed_paths.append(job_path) - for source_path in completed_paths: - del self._todo_dict[source_path] - - for job_path, job in jobs: # Wait the last jobs to finish - job.wait() - if job.returncode != 0: - error_paths.append((job_path, job.stdout.read() + job.stderr.read())) - if str(job_path) in self._hash_dict: - del self._hash_dict[str(job_path)] - else: - # Update hash if no error - for dependency_path in self._dependency_dict[str(job_path)]: - self._hash_dict[str(dependency_path)] = get_hash(dependency_path) - del self._todo_dict[job_path] - else: # Single-process - completed_paths: list[Path] = [] - for source_path, cmd in self._todo_dict.items(): - print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) - job = subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - universal_newlines=True, shell=True) - if job.returncode != 0: - error_paths.append((source_path, job.stdout + job.stderr)) - if str(source_path) in self._hash_dict: - del self._hash_dict[str(source_path)] + bin_path = config.BIN_DIR / app_path + if not bin_path.parent.exists(): + bin_path.parent.mkdir(parents=True) + object_files = [str(object_path) for _, object_path in compile_list + if object_path not in all_app_objects] + object_files.append(str(app_object_path)) + cmd = ' '.join([config.CC, config.COMMON_FLAGS, *object_files, '-o', str(bin_path), config.LINK_FLAGS]) + print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) + jobs.insert(0, (app_path, subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, + shell=True))) # FIFO style (will be poped) + if len(jobs) >= arguments.j: # If jobs count is maxed we wait for the oldest one to finished + app_path, oldest_job = jobs.pop() + oldest_job.wait() + if oldest_job.returncode != 0: + error_paths.append((source_path, oldest_job.stdout.read() + oldest_job.stderr.read())) break - # Update hash if no error - for dependency_path in self._dependency_dict[str(source_path)]: - self._hash_dict[str(dependency_path)] = get_hash(dependency_path) - completed_paths.append(source_path) - for source_path in completed_paths: - del self._todo_dict[source_path] + for job_path, job in jobs: # Wait the last jobs to finish + job.wait() + if job.returncode != 0: + error_paths.append((source_path, job.stdout.read() + job.stderr.read())) + else: # Single-process + for app_path, app_object_path in zip(config.APPS, all_app_objects): + if 'IGNORE_APPS' in config.__dict__ and app_path in config.IGNORE_APPS: + continue + bin_path = config.BIN_DIR / app_path + if not bin_path.parent.exists(): + bin_path.parent.mkdir(parents=True) + object_files = [str(object_path) for _, object_path in compile_list + if object_path not in all_app_objects] + object_files.append(str(app_object_path)) + cmd = ' '.join([config.CC, config.COMMON_FLAGS, *object_files, '-o', str(bin_path), config.LINK_FLAGS]) + print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) + job = subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True, shell=True) + if job.returncode != 0: + error_paths.append((source_path, job.stdout + job.stderr)) + if error_paths: + for error_path, error_text in error_paths: + print(ConsoleColor.RED + f'Error linking {error_path}:\n' + ConsoleColor.ENDCOLOR + error_text) + with open(os.path.join(config.OBJECT_DIR, 'hash.json'), 'w') as hash_file: + hash_file.write(json.dumps(hash_dict, indent=1)) + sys.exit(1) - if error_paths: - for error_path, error_text in error_paths: - print(ConsoleColor.RED + f'Error compiling {error_path}:\n' + ConsoleColor.ENDCOLOR + error_text) - with open(self._config.OBJECT_DIR / 'hash.json', 'w', encoding='utf-8') as hash_file: - hash_file.write(json.dumps(self._hash_dict, indent=1)) - return False - - # Running linking processes - if not self._config.BIN_DIR.exists(): - self._config.BIN_DIR.mkdir(parents=True) - all_app_objects = [self._config.OBJECT_DIR / Path(app_path).parent / (Path(app_path).stem + '.o') - for app_path in self._config.APPS] - if self._config.JOB_COUNT > 1: # Multi-process - jobs: list[tuple[Path, subprocess.Popen]] = [] - for app_path, app_object_path in zip(self._config.APPS, all_app_objects): - if app_path in self._config.IGNORE_APPS: - continue - bin_path = self._config.BIN_DIR / app_path - if not bin_path.parent.exists(): - bin_path.parent.mkdir(parents=True) - object_files = [str(object_path) for object_path in self._compile_dict.values() - if object_path not in all_app_objects] - object_files.append(str(app_object_path)) - cmd = ' '.join([self._config.CC, self._config.COMMON_FLAGS, *object_files, '-o', str(bin_path), - self._config.LINK_FLAGS]) - print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) - jobs.insert(0, (app_path, subprocess.Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, - shell=True))) # FIFO style (will be poped) - if len(jobs) >= self._config.JOB_COUNT: # If jobs count is maxed we wait for the oldest one to finished - app_path, oldest_job = jobs.pop() - oldest_job.wait() - if oldest_job.returncode != 0: - error_paths.append((app_path, oldest_job.stdout.read() + oldest_job.stderr.read())) - break - for job_path, job in jobs: # Wait the last jobs to finish - job.wait() - if job.returncode != 0: - error_paths.append((job_path, job.stdout.read() + job.stderr.read())) - else: # Single-process - for app_path, app_object_path in zip(self._config.APPS, all_app_objects): - if app_path in self._config.IGNORE_APPS: - continue - bin_path = self._config.BIN_DIR / app_path - if not bin_path.parent.exists(): - bin_path.parent.mkdir(parents=True) - object_files = [str(object_path) for object_path in self._compile_dict.values() - if object_path not in all_app_objects] - object_files.append(str(app_object_path)) - cmd = ' '.join([self._config.CC, self._config.COMMON_FLAGS, *object_files, '-o', str(bin_path), - self._config.LINK_FLAGS]) - print(ConsoleColor.BLUE + cmd + ConsoleColor.ENDCOLOR) - job = subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - universal_newlines=True, shell=True) - if job.returncode != 0: - error_paths.append((app_path, job.stdout + job.stderr)) - if error_paths: - for error_path, error_text in error_paths: - print(ConsoleColor.RED + f'Error linking {error_path}:\n' + ConsoleColor.ENDCOLOR + error_text) - with open(os.path.join(self._config.OBJECT_DIR, 'hash.json'), 'w', encoding='utf-8') as hash_file: - hash_file.write(json.dumps(self._hash_dict, indent=1)) - return False - - with open(self._config.OBJECT_DIR / 'hash.json', 'w', encoding='utf-8') as hash_file: - hash_file.write(json.dumps(self._hash_dict, indent=1)) - - print(f'{ConsoleColor.GREEN}Compilation done{ConsoleColor.ENDCOLOR}') - return True + with open(config.OBJECT_DIR / 'hash.json', 'w') as hash_file: + hash_file.write(json.dumps(hash_dict, indent=1)) diff --git a/watch.py b/watch.py deleted file mode 100644 index 81c49b6..0000000 --- a/watch.py +++ /dev/null @@ -1,151 +0,0 @@ -import ctypes -import ctypes.util -from enum import IntFlag -import os -from pathlib import Path -import select -from struct import unpack, calcsize -from typing import Callable - - -class WatchFlag(IntFlag): - # Supported events suitable for MASK parameter of INOTIFY_ADD_WATCH. - ACCESS = 0x00000001 # File was accessed - MODIFY = 0x00000002 # File was modified - ATTRIB = 0x00000004 # Metadata changed - CLOSE_WRITE = 0x00000008 # Writtable file was closed - CLOSE_NOWRITE = 0x00000010 # Unwrittable file closed - # CLOSE = (CLOSE_WRITE | CLOSE_NOWRITE) # Close - OPEN = 0x00000020 # File was opened - MOVED_FROM = 0x00000040 # File was moved from X - MOVED_TO = 0x00000080 # File was moved to Y - # MOVE = (MOVED_FROM | MOVED_TO) # Moves - CREATE = 0x00000100 # Subfile was created - DELETE = 0x00000200 # Subfile was deleted - DELETE_SELF = 0x00000400 # Self was deleted - MOVE_SELF = 0x00000800 # Self was moved - - # Events sent by the kernel. - UNMOUNT = 0x00002000 # Backing fs was unmounted - Q_OVERFLOW = 0x00004000 # Event queued overflowed - IGNORED = 0x00008000 # File was ignored - - # Helper events. - # CLOSE = (CLOSE_WRITE | CLOSE_NOWRITE) # Close - # MOVE = (MOVED_FROM | MOVED_TO) # Moves - - # Special flags. - ONLYDIR = 0x01000000 # Only watch the path if it is a directory - DONT_FOLLOW = 0x02000000 # Do not follow a sym link - EXCL_UNLINK = 0x04000000 # Exclude events on unlinked objects - MASK_CREATE = 0x10000000 # Only create watches - MASK_ADD = 0x20000000 # Add to the mask of an already existing watch - ISDIR = 0x40000000 # Event occurred against dir - ONESHOT = 0x80000000 # Only send event once - - # All events which a program can wait on. - # ALL_EVENTS = ( - # ACCESS | MODIFY | ATTRIB | CLOSE_WRITE | CLOSE_NOWRITE | OPEN | MOVED_FROM | MOVED_TO - # | CREATE | DELETE | DELETE_SELF | MOVE_SELF) - - -class InotifyError(Exception): - def __init__(self, message, *args, **kwargs): - message += f' ERRNO=({ctypes.get_errno()})' - super().__init__(message, *args, **kwargs) - - -WatcherCallback = Callable[[Path, WatchFlag], None] - - -class Watcher: - _EVENT_FORMAT = 'iIII' - _EVENT_SIZE = calcsize(_EVENT_FORMAT) - - def __init__(self): - libc_path = ctypes.util.find_library('c') - if libc_path is None: - libc_path = 'libc.so.6' - self._instance = ctypes.cdll.LoadLibrary(libc_path) - - self._inotify_init: ctypes.CDLL._FuncPtr = self._instance.inotify_init - self._inotify_init.argtypes = [] - self._inotify_init.restype = self._check_nonnegative - - self._inotify_add_watch: ctypes.CDLL._FuncPtr = self._instance.inotify_add_watch - self._inotify_add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32] - self._inotify_add_watch.restype = self._check_nonnegative - - self._inotify_rm_watch: ctypes.CDLL._FuncPtr = self._instance.inotify_rm_watch - self._inotify_rm_watch.argtypes = [ctypes.c_int, ctypes.c_int] - self._inotify_rm_watch.restype = self._check_nonnegative - - self._inotify_fd = self._inotify_init() - - self._poll = select.poll() - self._poll.register(self._inotify_fd, select.POLLIN) - - self._watch_fds: dict[Path, int] = {} - self._watch_info: dict[int, tuple[Path, WatcherCallback]] = {} - - def register(self, path: Path, flags: WatchFlag, callback: WatcherCallback): - if path in self._watch_fds: - self._inotify_rm_watch(self._inotify_fd, self._watch_fds[path]) - - watch_fd = self._inotify_add_watch(self._inotify_fd, str(path).encode(), flags) - self._watch_fds[path] = watch_fd - self._watch_info[watch_fd] = (path, callback) - - def unregister(self, path: Path): - watch_fd = self._watch_fds[path] - self._inotify_rm_watch(self._inotify_fd, watch_fd) - del self._watch_fds[path] - del self._watch_info[watch_fd] - - def watch(self): - while True: - results = self._poll.poll() - for poll_fd, _event in results: - watch_buffer = os.read(poll_fd, self._EVENT_SIZE) - if len(watch_buffer) < self._EVENT_SIZE: - continue - watch_fd, mask, _cookie, _namesize = unpack(self._EVENT_FORMAT, watch_buffer) - watch_path, callback = self._watch_info[watch_fd] - callback(watch_path, WatchFlag(mask)) - - def __del__(self): - self._poll.unregister(self._inotify_fd) - for _watch_path, watch_fd in self._watch_fds.items(): - self._inotify_rm_watch(self._inotify_fd, watch_fd) - os.close(self._inotify_fd) - - @staticmethod - def _check_nonnegative(result): - if result == -1: - raise InotifyError(f'Call failed (should not be -1): {result}') - return result - - -def main(): - from argparse import ArgumentParser # pylint: disable=import-outside-toplevel - - parser = ArgumentParser() - parser.add_argument('paths', nargs='*', type=Path) - arguments = parser.parse_args() - - paths: list[Path] = arguments.paths - - def callback(path: Path, _flags: WatchFlag): - print(f'{path} has been modified') - - watcher = Watcher() - for path in paths: - watcher.register(path, WatchFlag.MODIFY, callback) - try: - watcher.watch() - except KeyboardInterrupt: - print('\rExit') - - -if __name__ == '__main__': - main()