123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270 |
- # Script utility functions
- #
- # Copyright (C) 2014 Intel Corporation
- #
- # SPDX-License-Identifier: GPL-2.0-only
- #
- import argparse
- import glob
- import logging
- import os
- import random
- import shlex
- import shutil
- import string
- import subprocess
- import sys
- import tempfile
- import threading
- import importlib
- from importlib import machinery
- class KeepAliveStreamHandler(logging.StreamHandler):
- def __init__(self, keepalive=True, **kwargs):
- super().__init__(**kwargs)
- if keepalive is True:
- keepalive = 5000 # default timeout
- self._timeout = threading.Condition()
- self._stop = False
- # background thread waits on condition, if the condition does not
- # happen emit a keep alive message
- def thread():
- while not self._stop:
- with self._timeout:
- if not self._timeout.wait(keepalive):
- self.emit(logging.LogRecord("keepalive", logging.INFO,
- None, None, "Keepalive message", None, None))
- self._thread = threading.Thread(target = thread, daemon = True)
- self._thread.start()
- def close(self):
- # mark the thread to stop and notify it
- self._stop = True
- with self._timeout:
- self._timeout.notify()
- # wait for it to join
- self._thread.join()
- super().close()
- def emit(self, record):
- super().emit(record)
- # trigger timer reset
- with self._timeout:
- self._timeout.notify()
- def logger_create(name, stream=None, keepalive=None):
- logger = logging.getLogger(name)
- if keepalive is not None:
- loggerhandler = KeepAliveStreamHandler(stream=stream, keepalive=keepalive)
- else:
- loggerhandler = logging.StreamHandler(stream=stream)
- loggerhandler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
- logger.addHandler(loggerhandler)
- logger.setLevel(logging.INFO)
- return logger
- def logger_setup_color(logger, color='auto'):
- from bb.msg import BBLogFormatter
- for handler in logger.handlers:
- if (isinstance(handler, logging.StreamHandler) and
- isinstance(handler.formatter, BBLogFormatter)):
- if color == 'always' or (color == 'auto' and handler.stream.isatty()):
- handler.formatter.enable_color()
- def load_plugins(logger, plugins, pluginpath):
- import imp
- def load_plugin(name):
- logger.debug('Loading plugin %s' % name)
- spec = importlib.machinery.PathFinder.find_spec(name, path=[pluginpath] )
- if spec:
- return spec.loader.load_module()
- def plugin_name(filename):
- return os.path.splitext(os.path.basename(filename))[0]
- known_plugins = [plugin_name(p.__name__) for p in plugins]
- logger.debug('Loading plugins from %s...' % pluginpath)
- for fn in glob.glob(os.path.join(pluginpath, '*.py')):
- name = plugin_name(fn)
- if name != '__init__' and name not in known_plugins:
- plugin = load_plugin(name)
- if hasattr(plugin, 'plugin_init'):
- plugin.plugin_init(plugins)
- plugins.append(plugin)
- def git_convert_standalone_clone(repodir):
- """If specified directory is a git repository, ensure it's a standalone clone"""
- import bb.process
- if os.path.exists(os.path.join(repodir, '.git')):
- alternatesfile = os.path.join(repodir, '.git', 'objects', 'info', 'alternates')
- if os.path.exists(alternatesfile):
- # This will have been cloned with -s, so we need to convert it so none
- # of the contents is shared
- bb.process.run('git repack -a', cwd=repodir)
- os.remove(alternatesfile)
- def _get_temp_recipe_dir(d):
- # This is a little bit hacky but we need to find a place where we can put
- # the recipe so that bitbake can find it. We're going to delete it at the
- # end so it doesn't really matter where we put it.
- bbfiles = d.getVar('BBFILES').split()
- fetchrecipedir = None
- for pth in bbfiles:
- if pth.endswith('.bb'):
- pthdir = os.path.dirname(pth)
- if os.access(os.path.dirname(os.path.dirname(pthdir)), os.W_OK):
- fetchrecipedir = pthdir.replace('*', 'recipetool')
- if pthdir.endswith('workspace/recipes/*'):
- # Prefer the workspace
- break
- return fetchrecipedir
- class FetchUrlFailure(Exception):
- def __init__(self, url):
- self.url = url
- def __str__(self):
- return "Failed to fetch URL %s" % self.url
- def fetch_url(tinfoil, srcuri, srcrev, destdir, logger, preserve_tmp=False, mirrors=False):
- """
- Fetch the specified URL using normal do_fetch and do_unpack tasks, i.e.
- any dependencies that need to be satisfied in order to support the fetch
- operation will be taken care of
- """
- import bb
- checksums = {}
- fetchrecipepn = None
- # We need to put our temp directory under ${BASE_WORKDIR} otherwise
- # we may have problems with the recipe-specific sysroot population
- tmpparent = tinfoil.config_data.getVar('BASE_WORKDIR')
- bb.utils.mkdirhier(tmpparent)
- tmpdir = tempfile.mkdtemp(prefix='recipetool-', dir=tmpparent)
- try:
- tmpworkdir = os.path.join(tmpdir, 'work')
- logger.debug('fetch_url: temp dir is %s' % tmpdir)
- fetchrecipedir = _get_temp_recipe_dir(tinfoil.config_data)
- if not fetchrecipedir:
- logger.error('Searched BBFILES but unable to find a writeable place to put temporary recipe')
- sys.exit(1)
- fetchrecipe = None
- bb.utils.mkdirhier(fetchrecipedir)
- try:
- # Generate a dummy recipe so we can follow more or less normal paths
- # for do_fetch and do_unpack
- # I'd use tempfile functions here but underscores can be produced by that and those
- # aren't allowed in recipe file names except to separate the version
- rndstring = ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(8))
- fetchrecipe = os.path.join(fetchrecipedir, 'tmp-recipetool-%s.bb' % rndstring)
- fetchrecipepn = os.path.splitext(os.path.basename(fetchrecipe))[0]
- logger.debug('Generating initial recipe %s for fetching' % fetchrecipe)
- with open(fetchrecipe, 'w') as f:
- # We don't want to have to specify LIC_FILES_CHKSUM
- f.write('LICENSE = "CLOSED"\n')
- # We don't need the cross-compiler
- f.write('INHIBIT_DEFAULT_DEPS = "1"\n')
- # We don't have the checksums yet so we can't require them
- f.write('BB_STRICT_CHECKSUM = "ignore"\n')
- f.write('SRC_URI = "%s"\n' % srcuri)
- f.write('SRCREV = "%s"\n' % srcrev)
- f.write('WORKDIR = "%s"\n' % tmpworkdir)
- # Set S out of the way so it doesn't get created under the workdir
- f.write('S = "%s"\n' % os.path.join(tmpdir, 'emptysrc'))
- if not mirrors:
- # We do not need PREMIRRORS since we are almost certainly
- # fetching new source rather than something that has already
- # been fetched. Hence, we disable them by default.
- # However, we provide an option for users to enable it.
- f.write('PREMIRRORS = ""\n')
- f.write('MIRRORS = ""\n')
- logger.info('Fetching %s...' % srcuri)
- # FIXME this is too noisy at the moment
- # Parse recipes so our new recipe gets picked up
- tinfoil.parse_recipes()
- def eventhandler(event):
- if isinstance(event, bb.fetch2.MissingChecksumEvent):
- checksums.update(event.checksums)
- return True
- return False
- # Run the fetch + unpack tasks
- res = tinfoil.build_targets(fetchrecipepn,
- 'do_unpack',
- handle_events=True,
- extra_events=['bb.fetch2.MissingChecksumEvent'],
- event_callback=eventhandler)
- if not res:
- raise FetchUrlFailure(srcuri)
- # Remove unneeded directories
- rd = tinfoil.parse_recipe(fetchrecipepn)
- if rd:
- pathvars = ['T', 'RECIPE_SYSROOT', 'RECIPE_SYSROOT_NATIVE']
- for pathvar in pathvars:
- path = rd.getVar(pathvar)
- shutil.rmtree(path)
- finally:
- if fetchrecipe:
- try:
- os.remove(fetchrecipe)
- except FileNotFoundError:
- pass
- try:
- os.rmdir(fetchrecipedir)
- except OSError as e:
- import errno
- if e.errno != errno.ENOTEMPTY:
- raise
- bb.utils.mkdirhier(destdir)
- for fn in os.listdir(tmpworkdir):
- shutil.move(os.path.join(tmpworkdir, fn), destdir)
- finally:
- if not preserve_tmp:
- shutil.rmtree(tmpdir)
- tmpdir = None
- return checksums, tmpdir
- def run_editor(fn, logger=None):
- if isinstance(fn, str):
- files = [fn]
- else:
- files = fn
- editor = os.getenv('VISUAL', os.getenv('EDITOR', 'vi'))
- try:
- #print(shlex.split(editor) + files)
- return subprocess.check_call(shlex.split(editor) + files)
- except subprocess.CalledProcessError as exc:
- logger.error("Execution of '%s' failed: %s" % (editor, exc))
- return 1
- def is_src_url(param):
- """
- Check if a parameter is a URL and return True if so
- NOTE: be careful about changing this as it will influence how devtool/recipetool command line handling works
- """
- if not param:
- return False
- elif '://' in param:
- return True
- elif param.startswith('git@') or ('@' in param and param.endswith('.git')):
- return True
- return False
|