path: root/pym/portage/package/ebuild/_config/UseManager.py

# Copyright 2010-2012 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2

__all__ = (
	'UseManager',
)

from _emerge.Package import Package
from portage import os
from portage.dep import Atom, dep_getrepo, dep_getslot, ExtendedAtomDict, remove_slot, _get_useflag_re, _repo_separator
from portage.eapi import eapi_has_use_aliases, eapi_supports_stable_use_forcing_and_masking
from portage.exception import InvalidAtom
from portage.localization import _
from portage.util import grabfile, grabdict, grabdict_package, read_corresponding_eapi_file, stack_lists, writemsg
from portage.versions import _pkg_str

from portage.package.ebuild._config.helper import ordered_by_atom_specificity

class UseManager(object):

	def __init__(self, repositories, profiles, abs_user_config, is_stable,
		user_config=True):
		#	file				variable
		#--------------------------------
		#	repositories
		#--------------------------------
		#	use.mask			_repo_usemask_dict
		#	use.stable.mask			_repo_usestablemask_dict
		#	use.force			_repo_useforce_dict
		#	use.stable.force		_repo_usestableforce_dict
		#	use.aliases			_repo_usealiases_dict
		#	package.use.mask		_repo_pusemask_dict
		#	package.use.stable.mask		_repo_pusestablemask_dict
		#	package.use.force		_repo_puseforce_dict
		#	package.use.stable.force	_repo_pusestableforce_dict
		#	package.use.aliases		_repo_pusealiases_dict
		#--------------------------------
		#	profiles
		#--------------------------------
		#	use.mask			_usemask_list
		#	use.stable.mask			_usestablemask_list
		#	use.force			_useforce_list
		#	use.stable.force		_usestableforce_list
		#	package.use.mask		_pusemask_list
		#	package.use.stable.mask		_pusestablemask_list
		#	package.use			_pkgprofileuse
		#	package.use.force		_puseforce_list
		#	package.use.stable.force	_pusestableforce_list
		#--------------------------------
		#	user config
		#--------------------------------
		#	package.use			_pusedict

		# Dynamic variables tracked by the config class
		#--------------------------------
		#	profiles
		#--------------------------------
		#	usemask
		#	useforce
		#--------------------------------
		#	user config
		#--------------------------------
		#	puse

		self._user_config = user_config
		self._is_stable = is_stable
		self._repo_usemask_dict = self._parse_repository_files_to_dict_of_tuples("use.mask", repositories)
		self._repo_usestablemask_dict = \
			self._parse_repository_files_to_dict_of_tuples("use.stable.mask",
				repositories, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._repo_useforce_dict = self._parse_repository_files_to_dict_of_tuples("use.force", repositories)
		self._repo_usestableforce_dict = \
			self._parse_repository_files_to_dict_of_tuples("use.stable.force",
				repositories, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._repo_pusemask_dict = self._parse_repository_files_to_dict_of_dicts("package.use.mask", repositories)
		self._repo_pusestablemask_dict = \
			self._parse_repository_files_to_dict_of_dicts("package.use.stable.mask",
				repositories, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._repo_puseforce_dict = self._parse_repository_files_to_dict_of_dicts("package.use.force", repositories)
		self._repo_pusestableforce_dict = \
			self._parse_repository_files_to_dict_of_dicts("package.use.stable.force",
				repositories, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._repo_puse_dict = self._parse_repository_files_to_dict_of_dicts("package.use", repositories)

		self._usemask_list = self._parse_profile_files_to_tuple_of_tuples("use.mask", profiles)
		self._usestablemask_list = \
			self._parse_profile_files_to_tuple_of_tuples("use.stable.mask",
				profiles, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._useforce_list = self._parse_profile_files_to_tuple_of_tuples("use.force", profiles)
		self._usestableforce_list = \
			self._parse_profile_files_to_tuple_of_tuples("use.stable.force",
				profiles, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._pusemask_list = self._parse_profile_files_to_tuple_of_dicts("package.use.mask", profiles)
		self._pusestablemask_list = \
			self._parse_profile_files_to_tuple_of_dicts("package.use.stable.mask",
				profiles, eapi_filter=eapi_supports_stable_use_forcing_and_masking)
		self._pkgprofileuse = self._parse_profile_files_to_tuple_of_dicts("package.use", profiles, juststrings=True)
		self._puseforce_list = self._parse_profile_files_to_tuple_of_dicts("package.use.force", profiles)
		self._pusestableforce_list = \
			self._parse_profile_files_to_tuple_of_dicts("package.use.stable.force",
				profiles, eapi_filter=eapi_supports_stable_use_forcing_and_masking)

		self._pusedict = self._parse_user_files_to_extatomdict("package.use", abs_user_config, user_config)

		self._repo_usealiases_dict = self._parse_repository_usealiases(repositories)
		self._repo_pusealiases_dict = self._parse_repository_packageusealiases(repositories)

		self.repositories = repositories

	def _parse_file_to_tuple(self, file_name, recursive=True, eapi_filter=None):
		ret = []
		lines = grabfile(file_name, recursive=recursive)
		eapi = read_corresponding_eapi_file(file_name)
		if eapi_filter is not None and not eapi_filter(eapi):
			if lines:
				writemsg(_("--- EAPI '%s' does not support '%s': '%s'\n") %
					(eapi, os.path.basename(file_name), file_name),
					noiselevel=-1)
			return ()
		useflag_re = _get_useflag_re(eapi)
		for prefixed_useflag in lines:
			if prefixed_useflag[:1] == "-":
				useflag = prefixed_useflag[1:]
			else:
				useflag = prefixed_useflag
			if useflag_re.match(useflag) is None:
				writemsg(_("--- Invalid USE flag in '%s': '%s'\n") %
					(file_name, prefixed_useflag), noiselevel=-1)
			else:
				ret.append(prefixed_useflag)
		return tuple(ret)

	def _parse_file_to_dict(self, file_name, juststrings=False, recursive=True,
		eapi_filter=None, user_config=False):
		ret = {}
		location_dict = {}
		eapi = read_corresponding_eapi_file(file_name, default=None)
		if eapi is None and not user_config:
			eapi = "0"
		if eapi is None:
			ret = ExtendedAtomDict(dict)
		else:
			ret = {}
		file_dict = grabdict_package(file_name, recursive=recursive,
			allow_wildcard=(eapi is None), allow_repo=(eapi is None),
			verify_eapi=(eapi is not None))
		if eapi is not None and eapi_filter is not None and not eapi_filter(eapi):
			if file_dict:
				writemsg(_("--- EAPI '%s' does not support '%s': '%s'\n") %
					(eapi, os.path.basename(file_name), file_name),
					noiselevel=-1)
			return ret
		useflag_re = _get_useflag_re(eapi)
		for k, v in file_dict.items():
			useflags = []
			for prefixed_useflag in v:
				if prefixed_useflag[:1] == "-":
					useflag = prefixed_useflag[1:]
				else:
					useflag = prefixed_useflag
				if useflag_re.match(useflag) is None:
					writemsg(_("--- Invalid USE flag for '%s' in '%s': '%s'\n") %
						(k, file_name, prefixed_useflag), noiselevel=-1)
				else:
					useflags.append(prefixed_useflag)
			location_dict.setdefault(k, []).extend(useflags)
		for k, v in location_dict.items():
			if juststrings:
				v = " ".join(v)
			else:
				v = tuple(v)
			ret.setdefault(k.cp, {})[k] = v
		return ret

	def _parse_user_files_to_extatomdict(self, file_name, location, user_config):
		ret = ExtendedAtomDict(dict)
		if user_config:
			pusedict = grabdict_package(
				os.path.join(location, file_name), recursive=1, allow_wildcard=True, allow_repo=True, verify_eapi=False)
			for k, v in pusedict.items():
				ret.setdefault(k.cp, {})[k] = tuple(v)

		return ret

	def _parse_repository_files_to_dict_of_tuples(self, file_name, repositories, eapi_filter=None):
		ret = {}
		for repo in repositories.repos_with_profiles():
			ret[repo.name] = self._parse_file_to_tuple(os.path.join(repo.location, "profiles", file_name), eapi_filter=eapi_filter)
		return ret

	def _parse_repository_files_to_dict_of_dicts(self, file_name, repositories, eapi_filter=None):
		ret = {}
		for repo in repositories.repos_with_profiles():
			ret[repo.name] = self._parse_file_to_dict(os.path.join(repo.location, "profiles", file_name), eapi_filter=eapi_filter)
		return ret

	def _parse_profile_files_to_tuple_of_tuples(self, file_name, locations,
		eapi_filter=None):
		return tuple(self._parse_file_to_tuple(
			os.path.join(profile.location, file_name),
			recursive=profile.portage1_directories, eapi_filter=eapi_filter)
			for profile in locations)

	def _parse_profile_files_to_tuple_of_dicts(self, file_name, locations,
		juststrings=False, eapi_filter=None):
		return tuple(self._parse_file_to_dict(
			os.path.join(profile.location, file_name), juststrings,
			recursive=profile.portage1_directories, eapi_filter=eapi_filter,
			user_config=profile.user_config)
			for profile in locations)

	def _parse_repository_usealiases(self, repositories):
		ret = {}
		for repo in repositories.repos_with_profiles():
			file_name = os.path.join(repo.location, "profiles", "use.aliases")
			eapi = read_corresponding_eapi_file(file_name)
			useflag_re = _get_useflag_re(eapi)
			raw_file_dict = grabdict(file_name, recursive=True)
			file_dict = {}
			for real_flag, aliases in raw_file_dict.items():
				if useflag_re.match(real_flag) is None:
					writemsg(_("--- Invalid real USE flag in '%s': '%s'\n") % (file_name, real_flag), noiselevel=-1)
				else:
					for alias in aliases:
						if useflag_re.match(alias) is None:
							writemsg(_("--- Invalid USE flag alias for '%s' real USE flag in '%s': '%s'\n") %
								(real_flag, file_name, alias), noiselevel=-1)
						else:
							if any(alias in v for k, v in file_dict.items() if k != real_flag):
								writemsg(_("--- Duplicated USE flag alias in '%s': '%s'\n") %
									(file_name, alias), noiselevel=-1)
							else:
								file_dict.setdefault(real_flag, []).append(alias)
			ret[repo.name] = file_dict
		return ret

	def _parse_repository_packageusealiases(self, repositories):
		ret = {}
		for repo in repositories.repos_with_profiles():
			file_name = os.path.join(repo.location, "profiles", "package.use.aliases")
			eapi = read_corresponding_eapi_file(file_name)
			useflag_re = _get_useflag_re(eapi)
			lines = grabfile(file_name, recursive=True)
			file_dict = {}
			for line in lines:
				elements = line.split()
				atom = elements[0]
				try:
					atom = Atom(atom, eapi=eapi)
				except InvalidAtom:
					writemsg(_("--- Invalid atom in '%s': '%s'\n") % (file_name, atom))
					continue
				if len(elements) == 1:
					writemsg(_("--- Missing real USE flag for '%s' in '%s'\n") % (atom, file_name), noiselevel=-1)
					continue
				real_flag = elements[1]
				if useflag_re.match(real_flag) is None:
					writemsg(_("--- Invalid real USE flag for '%s' in '%s': '%s'\n") % (atom, file_name, real_flag), noiselevel=-1)
				else:
					for alias in elements[2:]:
						if useflag_re.match(alias) is None:
							writemsg(_("--- Invalid USE flag alias for '%s' real USE flag for '%s' in '%s': '%s'\n") %
								(real_flag, atom, file_name, alias), noiselevel=-1)
						else:
							# Duplicated USE flag aliases in entries for different atoms
							# matching the same package version are detected in getUseAliases().
							if any(alias in v for k, v in file_dict.get(atom.cp, {}).get(atom, {}).items() if k != real_flag):
								writemsg(_("--- Duplicated USE flag alias for '%s' in '%s': '%s'\n") %
									(atom, file_name, alias), noiselevel=-1)
							else:
								file_dict.setdefault(atom.cp, {}).setdefault(atom, {}).setdefault(real_flag, []).append(alias)
			ret[repo.name] = file_dict
		return ret

	def _isStable(self, pkg):
		if self._user_config:
			try:
				return pkg.stable
			except AttributeError:
				# KEYWORDS is unavailable (prior to "depend" phase)
				return False

		try:
			pkg._metadata
		except AttributeError:
			# KEYWORDS is unavailable (prior to "depend" phase)
			return False

		# Since repoman uses different config instances for
		# different profiles, we have to be careful to do the
		# stable check against the correct profile here.
		return self._is_stable(pkg)

	def getUseMask(self, pkg=None):
		if pkg is None:
			return frozenset(stack_lists(
				self._usemask_list, incremental=True))

		slot = None
		cp = getattr(pkg, "cp", None)
		if cp is None:
			slot = dep_getslot(pkg)
			repo = dep_getrepo(pkg)
			pkg = _pkg_str(remove_slot(pkg), slot=slot, repo=repo)
			cp = pkg.cp

		stable = self._isStable(pkg)

		usemask = []

		if hasattr(pkg, "repo") and pkg.repo != Package.UNKNOWN_REPO:
			repos = []
			try:
				repos.extend(repo.name for repo in
					self.repositories[pkg.repo].masters)
			except KeyError:
				pass
			repos.append(pkg.repo)
			for repo in repos:
				usemask.append(self._repo_usemask_dict.get(repo, {}))
				if stable:
					usemask.append(self._repo_usestablemask_dict.get(repo, {}))
				cpdict = self._repo_pusemask_dict.get(repo, {}).get(cp)
				if cpdict:
					pkg_usemask = ordered_by_atom_specificity(cpdict, pkg)
					if pkg_usemask:
						usemask.extend(pkg_usemask)
				if stable:
					cpdict = self._repo_pusestablemask_dict.get(repo, {}).get(cp)
					if cpdict:
						pkg_usemask = ordered_by_atom_specificity(cpdict, pkg)
						if pkg_usemask:
							usemask.extend(pkg_usemask)

		for i, pusemask_dict in enumerate(self._pusemask_list):
			if self._usemask_list[i]:
				usemask.append(self._usemask_list[i])
			if stable and self._usestablemask_list[i]:
				usemask.append(self._usestablemask_list[i])
			cpdict = pusemask_dict.get(cp)
			if cpdict:
				pkg_usemask = ordered_by_atom_specificity(cpdict, pkg)
				if pkg_usemask:
					usemask.extend(pkg_usemask)
			if stable:
				cpdict = self._pusestablemask_list[i].get(cp)
				if cpdict:
					pkg_usemask = ordered_by_atom_specificity(cpdict, pkg)
					if pkg_usemask:
						usemask.extend(pkg_usemask)

		return frozenset(stack_lists(usemask, incremental=True))

	def getUseForce(self, pkg=None):
		if pkg is None:
			return frozenset(stack_lists(
				self._useforce_list, incremental=True))

		cp = getattr(pkg, "cp", None)
		if cp is None:
			slot = dep_getslot(pkg)
			repo = dep_getrepo(pkg)
			pkg = _pkg_str(remove_slot(pkg), slot=slot, repo=repo)
			cp = pkg.cp

		stable = self._isStable(pkg)

		useforce = []

		if hasattr(pkg, "repo") and pkg.repo != Package.UNKNOWN_REPO:
			repos = []
			try:
				repos.extend(repo.name for repo in
					self.repositories[pkg.repo].masters)
			except KeyError:
				pass
			repos.append(pkg.repo)
			for repo in repos:
				useforce.append(self._repo_useforce_dict.get(repo, {}))
				if stable:
					useforce.append(self._repo_usestableforce_dict.get(repo, {}))
				cpdict = self._repo_puseforce_dict.get(repo, {}).get(cp)
				if cpdict:
					pkg_useforce = ordered_by_atom_specificity(cpdict, pkg)
					if pkg_useforce:
						useforce.extend(pkg_useforce)
				if stable:
					cpdict = self._repo_pusestableforce_dict.get(repo, {}).get(cp)
					if cpdict:
						pkg_useforce = ordered_by_atom_specificity(cpdict, pkg)
						if pkg_useforce:
							useforce.extend(pkg_useforce)

		for i, puseforce_dict in enumerate(self._puseforce_list):
			if self._useforce_list[i]:
				useforce.append(self._useforce_list[i])
			if stable and self._usestableforce_list[i]:
				useforce.append(self._usestableforce_list[i])
			cpdict = puseforce_dict.get(cp)
			if cpdict:
				pkg_useforce = ordered_by_atom_specificity(cpdict, pkg)
				if pkg_useforce:
					useforce.extend(pkg_useforce)
			if stable:
				cpdict = self._pusestableforce_list[i].get(cp)
				if cpdict:
					pkg_useforce = ordered_by_atom_specificity(cpdict, pkg)
					if pkg_useforce:
						useforce.extend(pkg_useforce)

		return frozenset(stack_lists(useforce, incremental=True))

	def getUseAliases(self, pkg):
		if not eapi_has_use_aliases(pkg.eapi):
			return {}

		cp = getattr(pkg, "cp", None)
		if cp is None:
			slot = dep_getslot(pkg)
			repo = dep_getrepo(pkg)
			pkg = _pkg_str(remove_slot(pkg), slot=slot, repo=repo)
			cp = pkg.cp

		usealiases = {}

		if hasattr(pkg, "repo") and pkg.repo != Package.UNKNOWN_REPO:
			repos = []
			try:
				repos.extend(repo.name for repo in
					self.repositories[pkg.repo].masters)
			except KeyError:
				pass
			repos.append(pkg.repo)
			for repo in repos:
				usealiases_dict = self._repo_usealiases_dict.get(repo, {})
				for real_flag, aliases in usealiases_dict.items():
					for alias in aliases:
						if any(alias in v for k, v in usealiases.items() if k != real_flag):
							writemsg(_("--- Duplicated USE flag alias for '%s%s%s': '%s'\n") %
								(pkg.cpv, _repo_separator, pkg.repo, alias), noiselevel=-1)
						else:
							usealiases.setdefault(real_flag, []).append(alias)
				cp_usealiases_dict = self._repo_pusealiases_dict.get(repo, {}).get(cp)
				if cp_usealiases_dict:
					usealiases_dict_list = ordered_by_atom_specificity(cp_usealiases_dict, pkg)
					for usealiases_dict in usealiases_dict_list:
						for real_flag, aliases in usealiases_dict.items():
							for alias in aliases:
								if any(alias in v for k, v in usealiases.items() if k != real_flag):
									writemsg(_("--- Duplicated USE flag alias for '%s%s%s': '%s'\n") %
										(pkg.cpv, _repo_separator, pkg.repo, alias), noiselevel=-1)
								else:
									usealiases.setdefault(real_flag, []).append(alias)

		return usealiases

	def getPUSE(self, pkg):
		cp = getattr(pkg, "cp", None)
		if cp is None:
			slot = dep_getslot(pkg)
			repo = dep_getrepo(pkg)
			pkg = _pkg_str(remove_slot(pkg), slot=slot, repo=repo)
			cp = pkg.cp
		ret = ""
		cpdict = self._pusedict.get(cp)
		if cpdict:
			puse_matches = ordered_by_atom_specificity(cpdict, pkg)
			if puse_matches:
				puse_list = []
				for x in puse_matches:
					puse_list.extend(x)
				ret = " ".join(puse_list)
		return ret

	def extract_global_USE_changes(self, old=""):
		ret = old
		cpdict = self._pusedict.get("*/*")
		if cpdict is not None:
			v = cpdict.pop("*/*", None)
			if v is not None:
				ret = " ".join(v)
				if old:
					ret = old + " " + ret
				if not cpdict:
					#No tokens left in atom_license_map, remove it.
					del self._pusedict["*/*"]
		return ret