bouquet/prototype/command.py

import os
import sys
import tomli
import tempfile
import subprocess
from dependency import *
from colorama import Fore, Style
from env import Environment
from path import Path
from config import Config


commands = {}


class CommandConfigurationError(Exception):
    pass


class CommandExecutionError(Exception):
    pass


class Command(DependencyItem):
    def all_commands():
        return commands


    def __init__(self, **kwargs):
        super().__init__()

        if 'name' in kwargs:
            self.__name = kwargs['name']
        if 'description' in kwargs:
            self.__description = kwargs['description']


    def name(self):
        return self.__name


    def description(self):
        return self.__description


    def dependencies(self):
        return None


    def register(cmd):
        commands[cmd.name()] = cmd


    def get(name):
        if name in commands:
            return commands[name]
        return None


    def run(self, args):
        return


class UserCommand(Command):
    def load_file(self, path):
        with open(path, mode='rb') as f:
            self.__data = tomli.load(f)


    def __init_dependencies(self):
        if 'dependency' not in self.__data:
            return

        dependency = self.__data['dependency']
        self.__dependencies = {}
        if 'seeds' in dependency:
            seeds = dependency['seeds']
            for s in seeds:
                self.__dependencies[s] = SeedDependency(parent=self.name(), name=s)

        if 'commands' in dependency:
            commands = dependency['commands']
            for c in commands:
                self.__dependencies[c] = ProgramDependency(parent=self.name(), name=c)


    def __init__(self, **kwargs):
        if 'path' not in kwargs:
            return

        self.__tmpdir = tempfile.TemporaryDirectory()
        self.__path = kwargs['path']
        self.load_file(self.__path)
        cmd = self.__data['command']
        super().__init__(name=cmd['name'], description=cmd['description'])

        self.__init_dependencies()


    def dependencies(self):
        return self.__dependencies


    def scan_directory(paths):
        for p in paths:
            for root, dirs, files in os.walk(p):
                for f in files:
                    if not f.endswith('.command'):
                        continue
                    cmd_path = os.path.join(root, f)
                    cmd = None
                    try:
                        cmd = UserCommand(path=cmd_path)
                    except tomli.TOMLDecodeError as e:
                        print('Err: Failed to parse command file {}\n     {}'.format(seed_path, e))
                        return -1

                    Command.register(cmd)


    def get_path_namespaces(self):
        ns = {
                'src': os.path.dirname(self.__path),
                'root-src': Environment.source_root(),
                'install': Environment.system_root(),
        }

        return ns


    def get_environment(self):
        return {
                'MEADOW_SYSROOT': Environment.system_root(),
                'MEADOW_NATIVE_ROOT': Environment.native_root(),
                'MEADOW_SOURCE_ROOT': Environment.source_root(),
                'MEADOW_TEMP': self.__tmpdir.name,
                'MEADOW_TARGET_PLATFORM': Config.get('target.platform'),
        }


    def run(self, args):
        from seed import Seed
        deps = self.resolve()

        for c in deps:
            build_system = None
            if isinstance(c, Seed):
                build_system = c.build_system()
                build_system.check_out_of_date()

        for c in deps:
            build_system = None
            if isinstance(c, Seed):
                build_system = c.build_system()

            if build_system is not None:
                out_of_date = \
                    build_system.is_configuration_out_of_date() \
                    or build_system.is_build_out_of_date()

                if out_of_date:
                    print(f'{Fore.GREEN}>>>{Style.RESET_ALL} Building {c.name()}')

                if build_system.is_configuration_out_of_date():
                    print(f'{Fore.CYAN}  >>{Style.RESET_ALL} Configuring build...')
                    build_system.configure(verbose=False)

                if out_of_date:
                    print(f'{Fore.CYAN}  >>{Style.RESET_ALL} Building...')
                    build_system.build(verbose=False)
                    print(f'{Fore.CYAN}  >>{Style.RESET_ALL} Installing...')
                    build_system.install(verbose=False)

        exec_info = self.__data.get('exec', None)
        if exec_info is None:
            raise CommandConfigurationError(f'Command {self.name()} has no executable configured')

        args = exec_info.get('args', None)
        if args is None or len(args) == 0:
            raise CommandConfigurationError(f'Command {self.name()} has no executable configured')

        ns = self.get_path_namespaces()
        expanded_args = []
        for p in args:
            parts = p.split(':')
            if len(parts) < 2 or parts[0] not in ns:
                expanded_args.append(p)
                continue

            paths = Path(p).expand_path(relative_to=ns)
            expanded_args.extend(paths)

        if expanded_args[0].endswith('.py'):
            expanded_args.insert(0, sys.executable)

        env = os.environ | self.get_environment()
        result = None

        try:
            result = subprocess.run(expanded_args, env=env)
        except KeyboardInterrupt:
            pass

        if result is not None and result.returncode != 0:
            raise CommandExecutionError(f'Execution of {self.name()} failed. See log for details')