# tinfoil: a simple wrapper around cooker for bitbake-based command-line utilities # # Copyright (C) 2012-2017 Intel Corporation # Copyright (C) 2011 Mentor Graphics Corporation # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License along # with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. import logging import os import sys import atexit import re from collections import OrderedDict, defaultdict import bb.cache import bb.cooker import bb.providers import bb.taskdata import bb.utils import bb.command import bb.remotedata from bb.cookerdata import CookerConfiguration, ConfigParameters from bb.main import setup_bitbake, BitBakeConfigParameters, BBMainException import bb.fetch2 # We need this in order to shut down the connection to the bitbake server, # otherwise the process will never properly exit _server_connections = [] def _terminate_connections(): for connection in _server_connections: connection.terminate() atexit.register(_terminate_connections) class TinfoilUIException(Exception): """Exception raised when the UI returns non-zero from its main function""" def __init__(self, returncode): self.returncode = returncode def __repr__(self): return 'UI module main returned %d' % self.returncode class TinfoilCommandFailed(Exception): """Exception raised when run_command fails""" class TinfoilDataStoreConnector: def __init__(self, tinfoil, dsindex): self.tinfoil = tinfoil self.dsindex = dsindex def getVar(self, name): value = self.tinfoil.run_command('dataStoreConnectorFindVar', self.dsindex, name) overrides = None if isinstance(value, dict): if '_connector_origtype' in value: value['_content'] = self.tinfoil._reconvert_type(value['_content'], value['_connector_origtype']) del value['_connector_origtype'] if '_connector_overrides' in value: overrides = value['_connector_overrides'] del value['_connector_overrides'] return value, overrides def getKeys(self): return set(self.tinfoil.run_command('dataStoreConnectorGetKeys', self.dsindex)) def getVarHistory(self, name): return self.tinfoil.run_command('dataStoreConnectorGetVarHistory', self.dsindex, name) def expandPythonRef(self, varname, expr, d): ds = bb.remotedata.RemoteDatastores.transmit_datastore(d) ret = self.tinfoil.run_command('dataStoreConnectorExpandPythonRef', ds, varname, expr) return ret def setVar(self, varname, value): if self.dsindex is None: self.tinfoil.run_command('setVariable', varname, value) else: # Not currently implemented - indicate that setting should # be redirected to local side return True def setVarFlag(self, varname, flagname, value): if self.dsindex is None: self.tinfoil.run_command('dataStoreConnectorSetVarFlag', self.dsindex, varname, flagname, value) else: # Not currently implemented - indicate that setting should # be redirected to local side return True def delVar(self, varname): if self.dsindex is None: self.tinfoil.run_command('dataStoreConnectorDelVar', self.dsindex, varname) else: # Not currently implemented - indicate that setting should # be redirected to local side return True def delVarFlag(self, varname, flagname): if self.dsindex is None: self.tinfoil.run_command('dataStoreConnectorDelVar', self.dsindex, varname, flagname) else: # Not currently implemented - indicate that setting should # be redirected to local side return True def renameVar(self, name, newname): if self.dsindex is None: self.tinfoil.run_command('dataStoreConnectorRenameVar', self.dsindex, name, newname) else: # Not currently implemented - indicate that setting should # be redirected to local side return True class TinfoilCookerAdapter: """ Provide an adapter for existing code that expects to access a cooker object via Tinfoil, since now Tinfoil is on the client side it no longer has direct access. """ class TinfoilCookerCollectionAdapter: """ cooker.collection adapter """ def __init__(self, tinfoil): self.tinfoil = tinfoil def get_file_appends(self, fn): return self.tinfoil.get_file_appends(fn) def __getattr__(self, name): if name == 'overlayed': return self.tinfoil.get_overlayed_recipes() elif name == 'bbappends': return self.tinfoil.run_command('getAllAppends') else: raise AttributeError("%s instance has no attribute '%s'" % (self.__class__.__name__, name)) class TinfoilRecipeCacheAdapter: """ cooker.recipecache adapter """ def __init__(self, tinfoil): self.tinfoil = tinfoil self._cache = {} def get_pkg_pn_fn(self): pkg_pn = defaultdict(list, self.tinfoil.run_command('getRecipes') or []) pkg_fn = {} for pn, fnlist in pkg_pn.items(): for fn in fnlist: pkg_fn[fn] = pn self._cache['pkg_pn'] = pkg_pn self._cache['pkg_fn'] = pkg_fn def __getattr__(self, name): # Grab these only when they are requested since they aren't always used if name in self._cache: return self._cache[name] elif name == 'pkg_pn': self.get_pkg_pn_fn() return self._cache[name] elif name == 'pkg_fn': self.get_pkg_pn_fn() return self._cache[name] elif name == 'deps': attrvalue = defaultdict(list, self.tinfoil.run_command('getRecipeDepends') or []) elif name == 'rundeps': attrvalue = defaultdict(lambda: defaultdict(list), self.tinfoil.run_command('getRuntimeDepends') or []) elif name == 'runrecs': attrvalue = defaultdict(lambda: defaultdict(list), self.tinfoil.run_command('getRuntimeRecommends') or []) elif name == 'pkg_pepvpr': attrvalue = self.tinfoil.run_command('getRecipeVersions') or {} elif name == 'inherits': attrvalue = self.tinfoil.run_command('getRecipeInherits') or {} elif name == 'bbfile_priority': attrvalue = self.tinfoil.run_command('getBbFilePriority') or {} elif name == 'pkg_dp': attrvalue = self.tinfoil.run_command('getDefaultPreference') or {} else: raise AttributeError("%s instance has no attribute '%s'" % (self.__class__.__name__, name)) self._cache[name] = attrvalue return attrvalue def __init__(self, tinfoil): self.tinfoil = tinfoil self.collection = self.TinfoilCookerCollectionAdapter(tinfoil) self.recipecaches = {} # FIXME all machines self.recipecaches[''] = self.TinfoilRecipeCacheAdapter(tinfoil) self._cache = {} def __getattr__(self, name): # Grab these only when they are requested since they aren't always used if name in self._cache: return self._cache[name] elif name == 'skiplist': attrvalue = self.tinfoil.get_skipped_recipes() elif name == 'bbfile_config_priorities': ret = self.tinfoil.run_command('getLayerPriorities') bbfile_config_priorities = [] for collection, pattern, regex, pri in ret: bbfile_config_priorities.append((collection, pattern, re.compile(regex), pri)) attrvalue = bbfile_config_priorities else: raise AttributeError("%s instance has no attribute '%s'" % (self.__class__.__name__, name)) self._cache[name] = attrvalue return attrvalue def findBestProvider(self, pn): return self.tinfoil.find_best_provider(pn) class Tinfoil: def __init__(self, output=sys.stdout, tracking=False, setup_logging=True): self.logger = logging.getLogger('BitBake') self.config_data = None self.cooker = None self.tracking = tracking self.ui_module = None self.server_connection = None if setup_logging: # This is the *client-side* logger, nothing to do with # logging messages from the server bb.msg.logger_create('BitBake', output) def __enter__(self): return self def __exit__(self, type, value, traceback): self.shutdown() def prepare(self, config_only=False, config_params=None, quiet=0): if self.tracking: extrafeatures = [bb.cooker.CookerFeatures.BASEDATASTORE_TRACKING] else: extrafeatures = [] if not config_params: config_params = TinfoilConfigParameters(config_only=config_only, quiet=quiet) cookerconfig = CookerConfiguration() cookerconfig.setConfigParameters(config_params) server, self.server_connection, ui_module = setup_bitbake(config_params, cookerconfig, extrafeatures) self.ui_module = ui_module # Ensure the path to bitbake's bin directory is in PATH so that things like # bitbake-worker can be run (usually this is the case, but it doesn't have to be) path = os.getenv('PATH').split(':') bitbakebinpath = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', 'bin')) for entry in path: if entry.endswith(os.sep): entry = entry[:-1] if os.path.abspath(entry) == bitbakebinpath: break else: path.insert(0, bitbakebinpath) os.environ['PATH'] = ':'.join(path) if self.server_connection: _server_connections.append(self.server_connection) if config_only: config_params.updateToServer(self.server_connection.connection, os.environ.copy()) self.run_command('parseConfiguration') else: self.run_actions(config_params) self.config_data = bb.data.init() connector = TinfoilDataStoreConnector(self, None) self.config_data.setVar('_remote_data', connector) self.cooker = TinfoilCookerAdapter(self) self.cooker_data = self.cooker.recipecaches[''] else: raise Exception('Failed to start bitbake server') def run_actions(self, config_params): """ Run the actions specified in config_params through the UI. """ ret = self.ui_module.main(self.server_connection.connection, self.server_connection.events, config_params) if ret: raise TinfoilUIException(ret) def parseRecipes(self): """ Force a parse of all recipes. Normally you should specify config_only=False when calling prepare() instead of using this function; this function is designed for situations where you need to initialise Tinfoil and use it with config_only=True first and then conditionally call this function to parse recipes later. """ config_params = TinfoilConfigParameters(config_only=False) self.run_actions(config_params) def run_command(self, command, *params): """ Run a command on the server (as implemented in bb.command). Note that there are two types of command - synchronous and asynchronous; in order to receive the results of asynchronous commands you will need to set an appropriate event mask using set_event_mask() and listen for the result using wait_event() - with the correct event mask you'll at least get bb.command.CommandCompleted and possibly other events before that depending on the command. """ if not self.server_connection: raise Exception('Not connected to server (did you call .prepare()?)') commandline = [command] if params: commandline.extend(params) result = self.server_connection.connection.runCommand(commandline) if result[1]: raise TinfoilCommandFailed(result[1]) return result[0] def set_event_mask(self, eventlist): """Set the event mask which will be applied within wait_event()""" if not self.server_connection: raise Exception('Not connected to server (did you call .prepare()?)') llevel, debug_domains = bb.msg.constructLogOptions() ret = self.run_command('setEventMask', self.server_connection.connection.getEventHandle(), llevel, debug_domains, eventlist) if not ret: raise Exception('setEventMask failed') def wait_event(self, timeout=0): """ Wait for an event from the server for the specified time. A timeout of 0 means don't wait if there are no events in the queue. Returns the next event in the queue or None if the timeout was reached. Note that in order to recieve any events you will first need to set the internal event mask using set_event_mask() (otherwise whatever event mask the UI set up will be in effect). """ if not self.server_connection: raise Exception('Not connected to server (did you call .prepare()?)') return self.server_connection.events.waitEvent(timeout) def get_overlayed_recipes(self): return defaultdict(list, self.run_command('getOverlayedRecipes')) def get_skipped_recipes(self): return OrderedDict(self.run_command('getSkippedRecipes')) def get_all_providers(self): return defaultdict(list, self.run_command('allProviders')) def find_providers(self): return self.run_command('findProviders') def find_best_provider(self, pn): return self.run_command('findBestProvider', pn) def get_runtime_providers(self, rdep): return self.run_command('getRuntimeProviders', rdep) def get_recipe_file(self, pn): """ Get the file name for the specified recipe/target. Raises bb.providers.NoProvider if there is no match or the recipe was skipped. """ best = self.find_best_provider(pn) if not best or (len(best) > 3 and not best[3]): skiplist = self.get_skipped_recipes() taskdata = bb.taskdata.TaskData(None, skiplist=skiplist) skipreasons = taskdata.get_reasons(pn) if skipreasons: raise bb.providers.NoProvider('%s is unavailable:\n %s' % (pn, ' \n'.join(skipreasons))) else: raise bb.providers.NoProvider('Unable to find any recipe file matching "%s"' % pn) return best[3] def get_file_appends(self, fn): return self.run_command('getFileAppends', fn) def parse_recipe(self, pn): """ Parse the specified recipe and return a datastore object representing the environment for the recipe. """ fn = self.get_recipe_file(pn) return self.parse_recipe_file(fn) def parse_recipe_file(self, fn, appends=True, appendlist=None, config_data=None): """ Parse the specified recipe file (with or without bbappends) and return a datastore object representing the environment for the recipe. Parameters: fn: recipe file to parse - can be a file path or virtual specification appends: True to apply bbappends, False otherwise appendlist: optional list of bbappend files to apply, if you want to filter them config_data: custom config datastore to use. NOTE: if you specify config_data then you cannot use a virtual specification for fn. """ if appends and appendlist == []: appends = False if config_data: dctr = bb.remotedata.RemoteDatastores.transmit_datastore(config_data) dscon = self.run_command('parseRecipeFile', fn, appends, appendlist, dctr) else: dscon = self.run_command('parseRecipeFile', fn, appends, appendlist) if dscon: return self._reconvert_type(dscon, 'DataStoreConnectionHandle') else: return None def build_file(self, buildfile, task): """ Runs the specified task for just a single recipe (i.e. no dependencies). This is equivalent to bitbake -b, except no warning will be printed. """ return self.run_command('buildFile', buildfile, task, True) def shutdown(self): if self.server_connection: self.run_command('clientComplete') _server_connections.remove(self.server_connection) bb.event.ui_queue = [] self.server_connection.terminate() self.server_connection = None def _reconvert_type(self, obj, origtypename): """ Convert an object back to the right type, in the case that marshalling has changed it (especially with xmlrpc) """ supported_types = { 'set': set, 'DataStoreConnectionHandle': bb.command.DataStoreConnectionHandle, } origtype = supported_types.get(origtypename, None) if origtype is None: raise Exception('Unsupported type "%s"' % origtypename) if type(obj) == origtype: newobj = obj elif isinstance(obj, dict): # New style class newobj = origtype() for k,v in obj.items(): setattr(newobj, k, v) else: # Assume we can coerce the type newobj = origtype(obj) if isinstance(newobj, bb.command.DataStoreConnectionHandle): connector = TinfoilDataStoreConnector(self, newobj.dsindex) newobj = bb.data.init() newobj.setVar('_remote_data', connector) return newobj class TinfoilConfigParameters(BitBakeConfigParameters): def __init__(self, config_only, **options): self.initial_options = options # Apply some sane defaults if not 'parse_only' in options: self.initial_options['parse_only'] = not config_only #if not 'status_only' in options: # self.initial_options['status_only'] = config_only if not 'ui' in options: self.initial_options['ui'] = 'knotty' if not 'argv' in options: self.initial_options['argv'] = [] super(TinfoilConfigParameters, self).__init__() def parseCommandLine(self, argv=None): # We don't want any parameters parsed from the command line opts = super(TinfoilConfigParameters, self).parseCommandLine([]) for key, val in self.initial_options.items(): setattr(opts[0], key, val) return opts