Manage the installation of external required or optional features.
It will browse the Features() to install, according to the OS of the computer. Must be sub-classed to create the appropriate Features(). Then, the installation is launched with:
>>> class SubInstaller(Installer):
>>> def __init__(self):
>>> super(SubInstaller, self).__init__()
>>> self._features = Features(req="", cmdos="")
>>> SubInstaller().install()
Create a new Installer instance.
def __init__(self):
"""Create a new Installer instance. """
self.__pbar = None
self.__pvalue = 0
self.__nb_steps = -1
self._features = None
self._python = self.__search_python_cmd()
logging.info(" ... the 'installer' is using the Python command: {:s}".format(self._python))
self.update_pip()
Set the progress bar.
def set_progress(self, progress):
"""Set the progress bar.
:param progress: (BaseProcessProgress) The installation progress.
"""
self.__pbar = progress
Return the list of feature identifiers.
def get_fids(self, feat_type=None):
"""Return the list of feature identifiers.
:param feat_type: (str) Only return features of the given type.
:returns: (list)
"""
if feat_type is None:
return self._features.get_ids()
f = list()
for fid in self._features.get_ids():
if self.feature_type(fid) == feat_type:
f.append(fid)
return f
Return the feature type: deps, lang, annot.
def feature_type(self, fid):
"""Return the feature type: deps, lang, annot.
:param fid: (str) Identifier of a feature
"""
ft = self._features.feature_type(fid)
return ft
Return True if the feature is enabled and/or set it.
def enable(self, fid, value=None):
"""Return True if the feature is enabled and/or set it.
:param fid: (str) Identifier of a feature
:param value: (bool or None) Enable of disable the feature.
"""
return self._features.enable(fid, value)
Return True if the feature is available and/or set it.
def available(self, fid, value=None):
"""Return True if the feature is available and/or set it.
:param fid: (str) Identifier of a feature
:param value: (bool or None) Make the feature available or not.
"""
return self._features.available(fid, value)
Return the long description of the feature.
def description(self, fid):
"""Return the long description of the feature.
:param fid: (str) Identifier of a feature
:return: (str)
"""
return self._features.description(fid)
Return the brief description of the feature.
def brief(self, fid):
"""Return the brief description of the feature.
:param fid: (str) Identifier of a feature
:return: (str)
"""
return self._features.brief(fid)
Update the version of pip.
def update_pip(self):
"""Update the version of pip.
"""
logging.info('Update pip, the package installer for Python:')
try:
process = sppasExecProcess()
process.run(self._python + ' -m pip install --upgrade pip --break-system-packages')
process.run(self._python + ' -m pip install --upgrade pip')
except:
try:
process = sppasExecProcess()
process.run(self._python + ' -m pip install --upgrade pip')
process.run(self._python + ' -m pip install --upgrade pip --break-system-packages')
except Exception as e:
raise sppasInstallationError(str(e))
Update the app config file for features depending on pip packages.
CAUTION: it is supposed that if the PIP dependency is satisfied, the feature can be enabled. It is currently True but it could be false...
sppasInstallationError
def ckeck_pypis(self):
"""Update the app config file for features depending on pip packages.
CAUTION: it is supposed that if the PIP dependency is satisfied, the
feature can be enabled. It is currently True but it could be false...
:param fid: (str) Identifier of a feature
:raises: sppasInstallationError
"""
for fid in self._features.get_ids('deps'):
for package, version in self._features.pypi(fid).items():
cfg.set_feature(fid, self._show_pypi(package))
cfg.save()
Launch the installation process.
None is used to install all feature types.
def install(self, feat_type=None):
"""Launch the installation process.
None is used to install all feature types.
:param feat_type: (str or None) Install features of the given type.
:return: (list) Error messages.
"""
try:
return self.__install(feat_type)
except Exception as e:
return str(e)
Install the given tar/zip file in the destination path of SPPAS.
@staticmethod
def install_resource(web_url, file_path, destination):
"""Install the given tar/zip file in the destination path of SPPAS.
:param web_url: (str) URL of the repo directory
:param file_path: (str) Tar/Zip filename to download and install
:param destination: (str) Destination path
"""
err, tmp_file = Installer.download_resource(web_url, file_path, destination)
if len(err) == 0:
if tmp_file.lower().endswith('zip') is True:
err = Installer.__unzip(tmp_file, destination)
elif tmp_file.lower().endswith('tar') is True:
err = Installer.__untar(tmp_file, destination)
else:
err = 'Unsupported file extension for file {:s}. Expected zip or tar.'.format(tmp_file)
if os.path.exists(tmp_file) is True and len(err) == 0:
os.remove(tmp_file)
if len(err) > 0:
raise sppasInstallationError(err)
Install the given tar/zip file in the destination path of SPPAS.
@staticmethod
def download_resource(web_url, file_path, destination):
"""Install the given tar/zip file in the destination path of SPPAS.
:param web_url: (str) URL of the repo directory
:param file_path: (str) Tar/Zip filename to download and install
:param destination: (str) Destination path
"""
url = web_url + file_path
url += '/download' if 'sourceforge' in web_url else ''
logging.debug('Downloading resource from: {}'.format(url))
logging.debug('Installing it to: {}'.format(destination))
tmp_file = os.path.join(destination, file_path)
err = Installer.__download(url, tmp_file)
return (err, tmp_file)
To be overridden. Return True if package is already installed.
def _search_package(self, package):
"""To be overridden. Return True if package is already installed.
:param package: (str) The system package to search.
"""
raise NotImplementedError
To be overridden. Install package.
def _install_package(self, package):
"""To be overridden. Install package.
:param package: (str) The system package to install.
:returns: False or None
"""
raise NotImplementedError
To be overridden. Return True if the package is up to date.
def _version_package(self, package, req_version):
"""To be overridden. Return True if the package is up to date.
:param package: (str) The system package to search.
:param req_version: (str) The minimum version required.
"""
raise NotImplementedError
To be overridden. Return True if the package need to be updated.
@staticmethod
def _need_update_package(stdout_show, req_version):
"""To be overridden. Return True if the package need to be updated.
:param stdout_show: (str) The stdout of the command.
:param req_version: (str) The minimum version required.
"""
raise NotImplementedError
To be overridden. Update package.
def _update_package(self, package, req_version):
"""To be overridden. Update package.
:param package: (str) The system package to update.
:returns: False or None
"""
raise NotImplementedError
Return True if given Pypi package is already installed.
def _show_pypi(self, package):
"""Return True if given Pypi package is already installed.
:param package: (str) The pip package to search.
"""
try:
command = self._python + ' -m pip show ' + package
process = sppasExecProcess()
process.run(command)
logging.debug('Search pypi return status: {}'.format(process.status()))
err = process.error()
stdout = process.out()
stdout = stdout.replace("b''", '')
if len(err) > 3 or len(stdout) == 0:
return False
except Exception as e:
raise sppasInstallationError(str(e))
return True
Return the pip command line to install or update a package.
def _pip_command(self, package, version, options='', args=''):
"""Return the pip command line to install or update a package.
:param package: (str) The pip package to install or update
:param version: (str) The version constraint
:param options: (str) The options of the package, like an index url
:param args: (str) The arguments of the pip install command
:return: (str) The command to be executed
"""
command = self._python + ' -m pip install'
if len(args) > 0:
command += ' ' + args
if len(options) > 0:
command += ' ' + options
command += " '" + package + version + "'"
if 'env' not in self._python:
command += ' --user'
command += ' --no-warn-script-location'
return command
Install a Python Pypi package.
sppasInstallationError
def _install_pypi(self, package, version, options=''):
"""Install a Python Pypi package.
:param package: (str) The pip package to install
:param version: (str) The version constraint
:raises: sppasInstallationError
:return: (str) Error message or empty string
"""
command = self._python + ' -m pip cache purge'
process = sppasExecProcess()
process.run(command)
logging.debug('python pip cache purge return status: {}'.format(process.status()))
logging.info('Try to install the stable version of {:s} python package.'.format(package))
command = self._pip_command(package, version, options, '--only-binary=:all:')
process = sppasExecProcess()
process.run(command)
err = u(process.error().strip())
stdout = process.out()
if len(stdout) > 3:
logging.info(stdout)
if len(err) > 0:
if len(options) > 0:
logging.warning('An error occurred when installing: {} {} (stable)'.format(package, version))
logging.warning("The stable package can't be installed from the official python package repository: {:s}.".format(str(err)))
err = self._install_prerelease_pypi(package, version, options)
else:
logging.error('The following error occurred: ' + err)
logging.info(" ... but we'll attempt another way to install the package:")
command = self._python + ' -m pip cache purge'
process = sppasExecProcess()
process.run(command)
logging.debug(' ... ... python pip cache purge return status: {}'.format(process.status()))
if len(stdout) > 3:
logging.info(stdout)
if len(err) > 0:
logging.error('The following error occurred: ' + err)
logging.info(" ... but we'll attempt a last way to install the package:")
command = self._pip_command(package, version, options)
process = sppasExecProcess()
process.run(command)
err = u(process.error().strip())
stdout = process.out()
if len(stdout) > 3:
logging.info(stdout)
if len(err) > 0:
logging.error('The following error occurred: ' + err)
if self._version_pypi(package, version) is False:
raise sppasInstallationError(err)
else:
return err
return ''
Install a pre-release version of a Python Pypi package.
A pre-release is an alpha or beta version, not already released in a stable package.
sppasInstallationError
def _install_prerelease_pypi(self, package, version, options=''):
"""Install a pre-release version of a Python Pypi package.
A pre-release is an alpha or beta version, not already released in a
stable package.
:param package: (str) The pip package to install
:param version: (str) The version constraint
:raises: sppasInstallationError
"""
command = self._python + ' -m pip cache purge'
process = sppasExecProcess()
process.run(command)
logging.debug('python pip cache purge return status: {}'.format(process.status()))
logging.info('Try to install a pre-build version of {:s} python package.'.format(package))
command = self._pip_command(package, version, options, '--pre')
process = sppasExecProcess()
process.run(command)
err = u(process.error().strip())
stdout = process.out()
if len(stdout) > 3:
logging.info(stdout)
if len(err) > 0:
if self._version_pypi(package, version) is False:
raise sppasInstallationError(err)
else:
return err
return ''
Returns True if package is up-to-date.
def _version_pypi(self, package, req_version):
"""Returns True if package is up-to-date.
:param package: (str) The pip package to search.
:param req_version: (str) The minimum version required.
"""
try:
command = self._python + ' -m pip show ' + package
process = sppasExecProcess()
process.run(command)
logging.debug('Version pypi return status: {}'.format(process.status()))
err = process.error()
out = process.out()
if len(err) > 3:
return False
return not Installer._need_update_pypi(out, req_version)
except:
return False
Return True if the package needs to be updated.
@staticmethod
def _need_update_pypi(stdout_show, req_version):
"""Return True if the package needs to be updated.
:param stdout_show: (str) The stdout of the command.
:param req_version: (str) The minimum version required.
"""
stdout_show = str(stdout_show)
req_version = str(req_version)
version = stdout_show.split('\\r\\n')[1].split(':')[1].replace(' ', '')
v = ''
i = 0
for letter in version:
if letter.isalpha() is False:
if letter == '.':
i += 1
if i == 2 or letter == ' ':
break
v += letter
else:
break
req_version = req_version.split(';', maxsplit=1)
comparator = req_version[0]
comparator += '='
v = v.strip()
v = float(v)
version = float(req_version[1])
if comparator == '>=':
return v < version
raise ValueError('The comparator: ' + comparator + ' does not refer to a valid comparator')
Update package.
sppasInstallationError
def _update_pypi(self, package, version, options=''):
"""Update package.
:param package: (str) The pip package to update.
:param version: (str) The version constraint
:param options: (str) The options of the package, like an index url
:raises: sppasInstallationError
"""
try:
command = self._python + ' -m pip cache purge'
process = sppasExecProcess()
process.run(command)
logging.debug('python pip cache purge return status: {}'.format(process.status()))
command = self._pip_command(package, version, options, '-U')
process = sppasExecProcess()
process.run(command)
logging.debug('Update pypi return status: {}'.format(process.status()))
except Exception as e:
raise sppasInstallationError(str(e))
err = u(process.error().strip())
stdout = u(process.out())
if len(stdout) > 3:
logging.info(stdout)
if len(err) > 0:
if self._version_pypi(package, version) is False:
raise sppasInstallationError(err)
else:
return err
return ''
Search for a valid python command. Raise SystemError if not found.
def __search_python_cmd(self):
"""Search for a valid python command. Raise SystemError if not found."""
logging.info("Search for a 'python' command that this installer can launch...")
logging.info(' ... Try: {:s}'.format(sys.executable))
command = quote(sys.executable)
command += " -c 'import sys; print(sys.version_info.major)' "
pyprocess = sppasExecProcess()
try:
pyprocess.run(command)
out = pyprocess.out().replace("b'", '')
out = out.replace("'", '')
out = out.strip()
logging.info('Command returned: {}'.format(out))
if len(out) == 1:
return quote(sys.executable)
except Exception as e:
logging.error(str(e))
command = "python3 -c 'import sys; print(sys.version_info.major)' "
logging.info(' ... Try: python3')
try:
pyprocess.run(command)
out = pyprocess.out().strip()
out = out.replace("b'", '')
out = out.replace("'", '')
out = out.strip()
logging.info('Command returned: {}'.format(out))
if len(out) == 1:
pyversion = int(out)
if pyversion == 3:
return 'python3'
except Exception as e:
logging.error(str(e))
command = "python -c 'import sys; print(sys.version_info.major)' "
logging.info(' ... Try: python')
try:
pyprocess.run(command)
out = pyprocess.out().replace("b'", '')
out = out.replace("'", '')
out = out.strip()
logging.info('Command returned: {}'.format(out))
if len(out) == 1:
pyversion = int(out)
if pyversion == 3:
return 'python'
except Exception as e:
logging.error(str(e))
raise SystemError('No valid python command can be invoked by the installer system.')
Really process the installation.
None is used to install all feature types.
def __install(self, feat_type=None):
"""Really process the installation.
None is used to install all feature types.
:param feat_type: (str or None) Install features of the given type.
:return: (list) Error messages.
"""
errors = list()
self.__pvalue = 1
self.__nb_steps = len(self._features.get_ids(feat_type))
for step, fid in enumerate(self._features.get_ids(feat_type)):
self.__pheader(self.__message('beginning_feature', fid))
if self._features.available(fid) is False:
pmesg = 'available_false'
elif self._features.enable(fid) is False:
if self._features.feature_type(fid) == 'deps':
if cfg.feature_installed(fid) is False:
cfg.set_feature(fid, False)
pmesg = 'enable_false'
else:
try:
fid_output = self.__install_feature(fid)
if len(fid_output) > 0:
msg = 'Installation of feature {} returned information message(s): '.format(fid)
errors.append(fid_output)
logging.warning(msg + fid_output)
except sppasInstallationError as e:
self._features.enable(fid, False)
pmesg = 'install_failed'
if self._features.feature_type(fid) == 'deps':
cfg.set_feature(fid, False)
errors.append(str(e))
logging.error(str(e))
except NotImplementedError:
self._features.available(fid, False)
self._features.enable(fid, False)
pmesg = 'install_failed'
msg = 'Installation of feature {} is not implemented yet for this os.'.format(fid)
errors.append(msg)
logging.error(msg)
except Exception as e:
logging.debug(traceback.format_exc())
self._features.enable(fid, False)
pmesg = 'install_failed'
msg = 'An un-expected error occurred: {}'.format(str(e))
errors.append(msg)
logging.error(msg)
else:
self._features.enable(fid, True)
cfg.set_feature(fid, True)
pmesg = 'install_success'
self.__pvalue = int(100.0 * float(step + 1) / self.__nb_steps)
self.__pupdate(pmesg, fid)
cfg.save()
return errors
Install the given feature depending on its type.
def __install_feature(self, fid):
"""Install the given feature depending on its type.
"""
ft = self._features.feature_type(fid)
out_msg = list()
if ft == 'deps':
if len(self._features.packages(fid)) > 0:
out_install = self.__install_packages(fid)
if len(out_install) > 0:
out_msg.append(out_install)
if len(self._features.pypi(fid)) > 0:
try:
out_install = self.__install_pypis(fid)
if len(out_install) > 0:
out_msg.append(out_install)
except sppasInstallationError as e:
try:
out_install = self.__install_pypis(fid, alt=True)
if len(out_install) > 0:
out_msg.append(out_install)
except sppasInstallationError as e_alt:
logging.error(str(e_alt))
raise sppasInstallationError(str(e))
else:
if self._features.check_feature_deps(fid) is False:
if len(self._features.pypi_alt(fid)) > 0:
out_install = self.__install_pypis(fid, alt=True)
if len(out_install) > 0:
out_msg.append(out_install)
if len(self._features.cmd(fid)) > 0:
out_install = self.__install_cmd(fid)
if len(out_install) > 0:
out_msg.append(out_install)
elif ft == 'lang':
self.__install_lang(fid)
elif ft == 'annot':
self.__install_annot(fid)
elif ft == 'spin':
self.__install_spin(fid)
else:
self._features.available(fid, False)
raise sppasInstallationError('Unknown feature type {}.'.format(fid))
return '\n\n'.join(out_msg)
Download, unzip and install resources for a given language.
sppasInstallationError
def __install_lang(self, fid):
"""Download, unzip and install resources for a given language.
:param fid: (str) Identifier of a feature
:raises: sppasInstallationError
"""
self.__pmessage('Download, unzip and install linguistic resources for {} language'.format(fid))
url = paths.url_lang_resources
if url.endswith('/') is False:
url += '/'
Installer.install_resource(url, self._features.lang(fid) + '.zip', paths.resources)
Download, unzip and install resources for a given annotation.
sppasInstallationError
def __install_annot(self, fid):
"""Download, unzip and install resources for a given annotation.
:param fid: (str) Identifier of a feature
:raises: sppasInstallationError
"""
self.__pmessage('Download, unzip and install resources for {} annotation'.format(fid))
url = paths.url_annot_resources
if url.endswith('/') is False:
url += '/'
Installer.install_resource(url, self._features.annot(fid) + '.tar', paths.resources)
Install a spin-off from its zip archive with targeted folders.
Expected archive structure: - sppas/... [expected] - resources/... [optional] Any other entry (file or folder) is ignored.
Behavior: - All entries under "sppas/" are extracted into the spin-off installation directory (paths.basedir or equivalent). - All entries under "resources/" are extracted into the global resources directory defined by paths.resources. - If the archive contains only "sppas/", it is installed normally. - No other top-level directories are supported.
sppasInstallationError
def __install_spin(self, fid):
"""Install a spin-off from its zip archive with targeted folders.
Expected archive structure:
- sppas/... [expected]
- resources/... [optional]
Any other entry (file or folder) is ignored.
Behavior:
- All entries under "sppas/" are extracted into the spin-off
installation directory (paths.basedir or equivalent).
- All entries under "resources/" are extracted into the global
resources directory defined by paths.resources.
- If the archive contains only "sppas/", it is installed normally.
- No other top-level directories are supported.
:param fid: (str) Identifier of the spin-off archive to install.
:raises: sppasInstallationError
"""
self.__pmessage('Download, unzip and install spin-off {}'.format(fid))
url = self._features.spin_url(fid)
if url.endswith('/') is False:
url += '/'
_file_path = ''.join([self._features.spin(fid), '-', self._features.spin_version(fid), '.zip'])
err, tmp_file = Installer.download_resource(url, _file_path, paths.basedir)
if len(err) > 0:
raise sppasInstallationError(err)
try:
with zipfile.ZipFile(tmp_file, 'r') as zf:
for info in zf.infolist():
name = info.filename
if name.endswith('/'):
continue
if name.startswith('resources/'):
target_root = paths.resources
relative_path = name[len('resources/'):]
elif name.startswith('sppas/'):
target_root = paths.sppas
relative_path = name[len('sppas/'):]
else:
logging.warning(f"Spin-off {fid}. Ignored entry because outside 'sppas/' or 'resources/': {name}")
continue
if relative_path == '':
continue
target_path = os.path.join(target_root, relative_path)
os.makedirs(os.path.dirname(target_path), exist_ok=True)
with zf.open(info, 'r') as src, open(target_path, 'wb') as dst:
dst.write(src.read())
except sppasInstallationError:
raise
except Exception as e:
logging.debug(traceback.format_exc())
raise sppasInstallationError(str(e))
if os.path.exists(tmp_file) is True:
os.remove(tmp_file)
Download the resource at given url and save into the tmp_file.
@staticmethod
def __download(url, tmp_file):
"""Download the resource at given url and save into the tmp_file.
:param url: (str) URL of the resource
:param tmp_file: (str) Filename of the temporary file to download into
:return: (str) Error message (if any)
"""
err = ''
try:
req = urllib.request.Request(url)
response = urllib.request.urlopen(req, timeout=60)
data = response.read()
except urllib.error.URLError as e:
logging.debug(traceback.format_exc())
if hasattr(e, 'reason'):
err = 'Failed to establish a connection to the url {}: {}'.format(url, e.reason)
elif hasattr(e, 'code'):
err = "The web server couldn't fulfill the request for url {}. Error code: {}".format(url, e.code)
else:
err = 'Unknown connection error.'
except http.client.IncompleteRead:
logging.debug(traceback.format_exc())
err = 'The server closed the connection before sending the complete file. The file can be downloaded at the following URL: {} and uncompressed in {}.'.format(url, paths.resources)
except Exception as e:
logging.debug(traceback.format_exc())
err = str(e)
else:
err = ''
try:
with open(tmp_file, 'wb') as out_file:
out_file.write(data)
except:
logging.debug(traceback.format_exc())
try:
zipped = response.read()
with open(tmp_file, 'wb') as f:
f.write(zipped)
except Exception as e:
logging.debug(traceback.format_exc())
err = str(e)
response.close()
if len(err) > 0 and REQUESTS_AVAILABLE is True:
chunk_size = 1024 * 1024
try:
response = requests.get(url, stream=True)
response.raise_for_status()
with open(tmp_file, 'wb') as out_file:
for chunk in response.iter_content(chunk_size=chunk_size):
out_file.write(chunk)
except requests.exceptions.RequestException as e:
err = f'Failed to download file using requests: {e}'
return err
@staticmethod
def __unzip(zip_filename, dest):
try:
z = zipfile.ZipFile(zip_filename)
z.extractall(dest)
z.close()
except zipfile.error as e:
logging.debug(traceback.format_exc())
return str(e)
return ''
@staticmethod
def __untar(tar_filename, dest):
try:
with tarfile.open(tar_filename, 'r') as tf:
tf.extractall(path=dest)
except Exception as e:
logging.debug(traceback.format_exc())
return str(e)
return ''
Execute a system command for a feature.
sppasInstallationError
def __install_cmd(self, fid):
"""Execute a system command for a feature.
:param fid: (str) Identifier of a feature
:raises: sppasInstallationError
"""
command = self._features.cmd(fid)
logging.info("The installer is testing the command '{}' for feature {}".format(command, fid))
try:
process = sppasExecProcess()
process.run(command)
err = u(process.error().strip())
stdout = process.out()
logging.info('Command return code is {}'.format(process.status()))
if len(stdout) > 3:
logging.info(stdout)
except Exception as e:
raise sppasInstallationError(str(e))
if process.status() != 0:
raise sppasInstallationError(err)
return err
Manage installation of system packages.
sppasInstallationError
def __install_packages(self, fid):
"""Manage installation of system packages.
:param fid: (str) Identifier of a feature
:raises: sppasInstallationError
"""
out_cmd = list()
for package, version in self._features.packages(fid).items():
logging.info(' ... package: {}'.format(package))
if self._search_package(package) is False:
logging.info(' ... ... is going to be installed.')
out_install = self._install_package(package)
if len(out_install) > 0:
out_cmd.append(out_install)
elif self._version_package(package, version) is False:
logging.info(' ... ... is going to be updated.')
try:
out_install = self._update_package(package, version)
if len(out_install) > 0:
out_cmd.append(out_install)
except Exception as e:
logging.error(' ... ... Update failed with the error: {}'.format(str(e)))
out_cmd.append(str(e))
else:
logging.info(' ... ... is already installed and up-to-date.')
self.__pvalue += self.__eval_step(fid) // self.__nb_steps
self.__pupdate('install_success', package + version)
return '\n'.join(out_cmd)
Manage the installation of pip packages.
sppasInstallationError
def __install_pypis(self, fid, alt=False):
"""Manage the installation of pip packages.
:param fid: (str) Identifier of a feature
:param alt: (bool) Install alternative pipys instead of the regular ones.
:raises: sppasInstallationError
"""
out_pip = list()
if alt is False:
pip_items = self._features.pypi(fid).items()
else:
pip_items = self._features.pypi_alt(fid).items()
if len(pip_items) == 0:
raise sppasInstallationError('No alternative pip packages found.')
for package, version in pip_items:
logging.info(' ... python library: {}'.format(package))
pip_options = self._features.pypi_opt(fid, package)
if self._show_pypi(package) is False:
try:
logging.info(' ... is going to be installed.')
err = self._install_pypi(package, version, pip_options)
if len(err) > 0:
out_pip.append(err)
except sppasInstallationError as e:
if 'DEPRECATION: ' in str(e):
logging.warning(' ... but returned a warning message. {}'.format(str(e)))
else:
logging.debug(' ... install_pypis re-raise the installation error')
raise sppasInstallationError(str(e))
elif self._version_pypi(package, version) is False:
try:
logging.info(' ... is going to be updated.')
self._update_pypi(package, version, pip_options)
except sppasInstallationError as e:
logging.error(' ... but the update failed with the error: {}'.format(str(e)))
out_pip.append(str(e))
else:
logging.info(' ... is already installed and up-to-date.')
time.sleep(2)
self.__pvalue += self.__eval_step(fid) // self.__nb_steps
if self._show_pypi(package) is True:
self.__pupdate('install_success', package + version)
else:
self.__pupdate('install_failed', package + version)
return '\n'.join(out_pip)
def __pheader(self, text):
if self.__pbar is not None:
self.__pbar.set_header(text)
self.__pbar.set_fraction(self.__pvalue)
self.__pbar.set_text('')
logging.info(' ==> {:d} percents'.format(self.__pvalue))
logging.info(' * * * * * {} * * * * * '.format(text))
def __pmessage(self, text):
if self.__pbar is not None:
self.__pbar.set_text(text)
logging.info(' ==> {text}'.format(text=text))
def __pfraction(self):
if self.__pbar is not None:
self.__pbar.set_fraction(self.__pvalue)
logging.info(' ==> {:d} percents'.format(self.__pvalue))
text is either a text or a mid.
def __pupdate(self, text, fid):
"""text is either a text or a mid."""
msg = self.__message(text, fid)
if self.__pvalue > 95:
self.__pvalue = 100
if self.__pbar is not None:
self.__pbar.update(self.__pvalue, msg)
logging.info(' ==> {:s} ({:d} percents)'.format(str(msg), self.__pvalue))
def __message(self, mid, fid):
if mid in MESSAGES:
return MESSAGES[mid].format(name=fid)
else:
return mid
Return the percentage of 1 step in progression for a given feature.
def __eval_step(self, fid):
"""Return the percentage of 1 step in progression for a given feature.
:param fid: (str) Identifier of a feature
:return: (float)
"""
nb_total = 0
ft = self._features.feature_type(fid)
if ft == 'deps':
nb_cmd = 0
if len(self._features.cmd(fid)) > 0:
nb_cmd = 1
nb_packages = len(self._features.packages(fid))
nb_pypi = len(self._features.pypi(fid))
nb_total = nb_cmd + nb_packages + nb_pypi
elif ft == 'annot':
if len(self._features.annot(fid)) > 0:
nb_total = 1
elif ft == 'lang':
if len(self._features.lang(fid)) > 0:
nb_total = 1
if nb_total > 0:
return int(round(1.0 / float(nb_total), 2) * 100.0)
return 0