Source code for spack.tag

# Copyright 2013-2024 Lawrence Livermore National Security, LLC and other
# Spack Project Developers. See the top-level COPYRIGHT file for details.
#
# SPDX-License-Identifier: (Apache-2.0 OR MIT)
"""Classes and functions to manage package tags"""
import collections
import copy
from collections.abc import Mapping

import spack.error
import spack.util.spack_json as sjson


def _get_installed_package_names():
    """Returns names of packages installed in the active environment."""
    specs = spack.environment.installed_specs()
    return [spec.name for spec in specs]




def packages_with_tags(tags, installed, skip_empty):
    """
    Returns a dict, indexed by tag, containing lists of names of packages
    containing the tag or, if no tags, for all available tags.

    Arguments:
        tags (list or None): list of tags of interest or None for all
        installed (bool): True if want names of packages that are installed;
            otherwise, False if want all packages with the tag
        skip_empty (bool): True if exclude tags with no associated packages;
            otherwise, False if want entries for all tags even when no such
            tagged packages
    """
    tag_pkgs = collections.defaultdict(lambda: list)
    spec_names = _get_installed_package_names() if installed else []
    keys = spack.repo.PATH.tag_index if tags is None else tags
    for tag in keys:
        packages = [
            name for name in spack.repo.PATH.tag_index[tag] if not installed or name in spec_names
        ]
        if packages or not skip_empty:
            tag_pkgs[tag] = packages
    return tag_pkgs





class TagIndex(Mapping):
    """Maps tags to list of packages."""

    def __init__(self, repository):
        self._tag_dict = collections.defaultdict(list)
        self.repository = repository

    @property
    def tags(self):
        return self._tag_dict



    def to_json(self, stream):
        sjson.dump({"tags": self._tag_dict}, stream)




    @staticmethod
    def from_json(stream, repository):
        d = sjson.load(stream)

        if not isinstance(d, dict):
            raise TagIndexError("TagIndex data was not a dict.")

        if "tags" not in d:
            raise TagIndexError("TagIndex data does not start with 'tags'")

        r = TagIndex(repository=repository)

        for tag, packages in d["tags"].items():
            r[tag].extend(packages)

        return r


    def __getitem__(self, item):
        return self._tag_dict[item]

    def __iter__(self):
        return iter(self._tag_dict)

    def __len__(self):
        return len(self._tag_dict)



    def copy(self):
        """Return a deep copy of this index."""
        clone = TagIndex(repository=self.repository)
        clone._tag_dict = copy.deepcopy(self._tag_dict)
        return clone




    def get_packages(self, tag):
        """Returns all packages associated with the tag."""
        return self.tags[tag] if tag in self.tags else []




    def merge(self, other):
        """Merge another tag index into this one.

        Args:
            other (TagIndex): tag index to be merged
        """
        other = other.copy()  # defensive copy.

        for tag in other.tags:
            if tag not in self.tags:
                self.tags[tag] = other.tags[tag]
                continue

            spkgs, opkgs = self.tags[tag], other.tags[tag]
            self.tags[tag] = sorted(list(set(spkgs + opkgs)))




    def update_package(self, pkg_name):
        """Updates a package in the tag index.

        Args:
            pkg_name (str): name of the package to be removed from the index
        """
        pkg_cls = self.repository.get_pkg_class(pkg_name)

        # Remove the package from the list of packages, if present
        for pkg_list in self._tag_dict.values():
            if pkg_name in pkg_list:
                pkg_list.remove(pkg_name)

        # Add it again under the appropriate tags
        for tag in getattr(pkg_cls, "tags", []):
            tag = tag.lower()
            self._tag_dict[tag].append(pkg_cls.name)






class TagIndexError(spack.error.SpackError):
    """Raised when there is a problem with a TagIndex."""