Fix watch if compilation fails, update README

* Fix issue #1

README.md

@@ -10,6 +10,8 @@ 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
 
@@ -25,7 +27,13 @@ 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`.
 
-**Be careful** : `clean` will delete the object and binary output directories defined in `make.py`.
+**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!
 
 
 ## How it works
@@ -35,19 +43,13 @@ By default 3 targets can be used `make` (or `make all` as usual default of makef
 µ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 objects files is meant to switch mode quickly (only binaries are rebuilt).
+The separation of object/release objects files is meant to avoid rebuilding the whole project after debugging a single file.
 
 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.

make.py

@@ -8,11 +8,12 @@ from umake import make
 
 class Config:
     CC = 'g++'                             # Compiler to call
-    APPS = ['app_name']                    # Output binaries (need to be found as .cpp directly in SOURCE_DIR)
+    APPS = ['app_name']                    # Output binaries (path to source without extension)
     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
@@ -23,8 +24,9 @@ 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`)
 
-    CPP_HEADERS = INCLUDE_DIR.rglob('*.hpp')
-    CPP_SOURCES = SOURCE_DIR.rglob('*.cpp')
+    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('.')]
 
 
 def main():

makefile

@@ -1,13 +1,13 @@
 MAKE_PATH="."
 UMAKE_PATH="umake"
 
-.PHONY: all clean
+.PHONY: all debug clean
 
 all:
-	@PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py
+	@PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py
 
 debug:
-	@PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py --type debug
+	@PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py --type debug
 
 clean:
-	@PYTHONPATH=$(UMAKE_PATH) python $(MAKE_PATH)/make.py --clean
+	@PYTHONPATH=$(UMAKE_PATH) python3 $(MAKE_PATH)/make.py --clean

setup.cfg

@@ -0,0 +1,25 @@
+[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

umake.py

@@ -13,7 +13,9 @@ 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)
@@ -26,12 +28,14 @@ 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`)
 
-    CPP_HEADERS = INCLUDE_DIR.rglob('*.hpp')
+    PRE_COMPILE_FUNCTION = None  # Function to run before compile (not call in --clean situation)
+
     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'
@@ -43,8 +47,8 @@ class ConsoleColor:
 
 
 def get_hash(path: Path) -> str:
-    with open(path, 'r') as hashing_file:
-        hash_obj = hashlib.md5()
+    hash_obj = hashlib.md5()
+    with open(path, 'r', encoding='utf-8') as hashing_file:
         hash_obj.update(hashing_file.read().encode())
     return hash_obj.hexdigest()
 
@@ -58,10 +62,19 @@ def make(config: Config):
 
     # Clean action
     if arguments.clean:
-        shutil.rmtree(config.OBJECT_DIR, ignore_errors=True)
-        shutil.rmtree(config.BIN_DIR, ignore_errors=True)
+        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)
         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
@@ -70,158 +83,223 @@ def make(config: Config):
         config.COMMON_FLAGS += ' ' + config.COMMON_RELEASE_FLAGS
         config.OBJECT_DIR = config.OBJECT_DIR / 'release'
 
-    # 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]] = []
+    # Update job count
+    config.JOB_COUNT = arguments.j
 
-    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())
+    Builder(config).make()
 
-    # 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)
 
-    # 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,
+class Builder:
+    def __init__(self, config: Config):
+        self._config = config
+
+        # 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] = {}
+
+        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'])
+            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)
+
+    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)
+                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)])
-        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
 
-    if not todo_list:
-        print(ConsoleColor.GREEN + 'Nothing to do' + ConsoleColor.ENDCOLOR)
-        return
+    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(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()))
+        # 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)]
                     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
-            # 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)
+                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]
 
-    # 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
-            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
-        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
 
-    with open(config.OBJECT_DIR / 'hash.json', 'w') as hash_file:
-        hash_file.write(json.dumps(hash_dict, indent=1))
+        # 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

watch.py

@@ -0,0 +1,151 @@
+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()