ninja_watch

#!/usr/bin/env python3
"""ninja-watch is a tool to wait for changes to dependencies in ninja files"""

from typing import Dict, List, Sequence, TypeVar

import sys
import os
import argparse
import subprocess
import time
import re

T = TypeVar('T')

LANG_EXT_MAP: Dict[str, List] = {
    'c': ['.c', '.h'],
    'cpp': ['.h', '.cpp'],
    'rust': ['.rs']
}

LANG_SFILES_MAP: Dict[str, List] = {
    'c': [],
    'cpp': [],
    'rust': ['Cargo.toml']
}


def eprint(*args):
    print('ninja-watch: {}'.format(*args), file=sys.stderr)


def flatten(container: Sequence[List[T]]) -> List[T]:
    res: List[T] = []
    for item in container:
        res += item
    return res


def get_fileexts(langs):
    file_exts = [LANG_EXT_MAP[x] for x in langs]
    file_exts = flatten(file_exts)
    file_exts = list(set(file_exts))  # make unique
    return file_exts


def get_special_files(root, langs):
    tmp = [LANG_SFILES_MAP[x] for x in langs]
    filenames = flatten(tmp)
    filenames += ['meson.build']
    res = []
    for walk_root, _, files in os.walk(root):
        for fname in files:
            if any([fname == filename for filename in filenames]):
                res.append(os.path.join(walk_root, fname))
    return res


def get_metadata(buildroot):
    """Get languages from configure system (mesonbuild...)"""
    with open(os.path.join(buildroot, 'build.ninja')) as ninjafile:
        ninja = ninjafile.read().strip()
        match = re.search(r'regenerate[\s]*([^\s]*)', ninja)
        if match:
            srcroot = match.group(1)
            mesonfilename = os.path.join(srcroot, 'meson.build')
        else:
            return ('', [])
        with open(mesonfilename) as mesonfile:
            meson = mesonfile.read().strip()
            project_regex = r'project\([^,]*,[\s]*((\'[^\']*\'[\s,]*)+)'
            match = re.search(project_regex, meson)
            if match:
                langs = match.group(1).strip(',').split(',')
                langs = [x.strip().strip('\'') for x in langs]
                langs = [x for x in langs if x != '']
                return (srcroot, langs)
    return ('', [])


def get_list(ninja_args, buildroot):
    """Get list of files"""
    try:
        deps = subprocess.check_output(['ninja'] + ninja_args)
        deps = deps.decode('utf-8').strip().split('\n')
    except subprocess.CalledProcessError:
        return []
    (srcroot, langs) = get_metadata(buildroot)
    exts = get_fileexts(langs)
    special_files = get_special_files(srcroot, langs)
    deps = [x.strip() for x in deps]
    deps = [x for x in deps if any([x.endswith(ext) for ext in exts])]
    deps = [os.path.abspath(os.path.join(buildroot, x)) for x in deps]
    deps += special_files
    deps = list(set(deps))  # Make unique

    dirs = [os.path.dirname(x) for x in deps]
    dirs = list(set(dirs))  # Make unique
    return deps + dirs


def main(argv):
    """Main function"""
    parser = argparse.ArgumentParser(
        description='''Rerun ninja as soon as files are modified or created.
        All arguments (except -h) will be passed through to ninja.'''
    )
    parser.add_argument(
        '-C',
        nargs='?',
        help='Which directory to run `ninja-build -t deps` in'
    )
    (args, _) = parser.parse_known_args(argv)

    ninja_args = ['-t', 'deps']
    if args.C:
        ninja_args = ['-C', args.C] + ninja_args

    while True:
        subprocess.call(['ninja'] + sys.argv[1:])
        deps = get_list(ninja_args, args.C if args.C else '.')
        if deps:
            output = subprocess.check_output(
                [
                    'inotifywait',
                    '-q',
                    '--format', '%e: %w %f',
                    '-e', 'modify'
                ] + deps
            )
            eprint("Triggered by {}".format(output.decode('utf-8')))
            time.sleep(0.5)
        else:
            eprint("fatal: could not find any sources")
            time.sleep(5)


if __name__ == '__main__':
    eprint('Press CTRL+C twice to exit')
    while True:
        try:
            main(sys.argv[1:])
        except KeyboardInterrupt:
            eprint('Rebuilding. If you want to quit press CTRL+C again')
            try:
                time.sleep(1)
            except KeyboardInterrupt:
                break