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')