cloud_cache.py

from typing import List, Tuple, Dict, Optional, Union
from abc import ABC, abstractmethod
from pathlib import Path
import os
import pathlib
import pandas as pd
import boto3
import semver
import tqdm
import re
import json
import warnings
from botocore import UNSIGNED
from botocore.client import Config
from allensdk.internal.core.lims_utilities import safe_system_path
from allensdk.api.cloud_cache.manifest import Manifest
from allensdk.api.cloud_cache.file_attributes import CacheFileAttributes
from allensdk.api.cloud_cache.utils import file_hash_from_path
from allensdk.api.cloud_cache.utils import bucket_name_from_url
from allensdk.api.cloud_cache.utils import relative_path_from_url


class OutdatedManifestWarning(UserWarning):
    pass


class MissingLocalManifestWarning(UserWarning):
    pass


class BasicLocalCache(ABC):
    """
    A class to handle the loading and accessing a project's data and
    metadata from a local cache directory. Does NOT include any 'smart'
    features like:
    1. Keeping track of last loaded manifest
    2. Constructing symlinks for valid data from previous dataset versions
    3. Warning of outdated manifests

    For those features (and more) see the CloudCacheBase class

    Parameters
    ----------
    cache_dir: str or pathlib.Path
        Path to the directory where data and metadata are stored on the
        local system

    project_name: str
        the name of the project this cache is supposed to access. This will
        be the root directory for all files stored in the bucket.

    ui_class_name: Optional[str]
        Name of the class users are actually using to manipulate this
        functionality (used to populate helpful error messages)
    """

    def __init__(
        self,
        cache_dir: Union[str, Path],
        project_name: str,
        ui_class_name: Optional[str] = None
    ):
        os.makedirs(cache_dir, exist_ok=True)

        # the class users are actually interacting with
        # (for warning message purposes)
        if ui_class_name is None:
            self._user_interface_class = type(self).__name__
        else:
            self._user_interface_class = ui_class_name

        self._manifest = None
        self._manifest_name = None

        self._cache_dir = cache_dir
        self._project_name = project_name

        self._manifest_file_names = self._list_all_manifests()

    # ====================== BasicLocalCache properties =======================

    @property
    def ui(self):
        return self._user_interface_class

    @property
    def current_manifest(self) -> Union[None, str]:
        """The name of the currently loaded manifest"""
        return self._manifest_name

    @property
    def project_name(self) -> str:
        """The name of the project that this cache is accessing"""
        return self._project_name

    @property
    def manifest_prefix(self) -> str:
        """On-line prefix for manifest files"""
        return f'{self.project_name}/manifests/'

    @property
    def file_id_column(self) -> str:
        """The col name in metadata files used to uniquely identify data files
        """
        return self._manifest.file_id_column

    @property
    def version(self) -> str:
        """The version of the dataset currently loaded"""
        return self._manifest.version

    @property
    def metadata_file_names(self) -> list:
        """List of metadata file names associated with this dataset"""
        return self._manifest.metadata_file_names

    @property
    def manifest_file_names(self) -> list:
        """Sorted list of manifest file names associated with this dataset
        """
        return self._manifest_file_names

    @property
    def latest_manifest_file(self) -> str:
        """parses on-line available manifest files for semver string
        and returns the latest one
        self.manifest_file_names are assumed to be of the form
        '<anything>_v<semver_str>.json'

        Returns
        -------
        str
            the filename whose semver string is the latest one
        """
        return self._find_latest_file(self.manifest_file_names)

    @property
    def cache_dir(self) -> str:
        """Return the cache directory path.

        Returns
        -------
        str
            Full cache directory path
        """
        return self._cache_dir

    # ====================== BasicLocalCache methods ==========================

    @abstractmethod
    def _list_all_manifests(self) -> list:
        """
        Return a list of all of the file names of the manifests associated
        with this dataset
        """
        raise NotImplementedError()

    def list_all_downloaded_manifests(self) -> list:
        """
        Return a list of all of the manifest files that have been
        downloaded for this dataset
        """
        output = [x for x in os.listdir(self._cache_dir)
                  if re.fullmatch(".*_manifest_v.*.json", x)]
        output.sort()
        return output

    def _find_latest_file(self, file_name_list: List[str]) -> str:
        """
        Take a list of files named like

        {blob}_v{version}.json

        and return the one with the latest version
        """
        vstrs = [s.split(".json")[0].split("_v")[-1]
                 for s in file_name_list]
        versions = [semver.VersionInfo.parse(v) for v in vstrs]
        imax = versions.index(max(versions))
        return file_name_list[imax]

    def _load_manifest(
        self,
        manifest_name: str,
        use_static_project_dir: bool = False
    ) -> Manifest:
        """
        Load and return a manifest from this dataset.

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        use_static_project_dir: bool
            When determining what the local path of a remote resource
            (data or metadata file) should be, the Manifest class will
            typically create a versioned project subdirectory under the user
            provided `cache_dir`
            (e.g. f"{cache_dir}/{project_name}-{manifest_version}")
            to allow the possibility of multiple manifest (and data) versions
            to be used. In certain cases, like when using a project's s3 bucket
            directly as the cache_dir, the project directory name needs to be
            static (e.g. f"{cache_dir}/{project_name}"). When set to True,
            the Manifest class will use a static project directory to determine
            local paths for remote resources. Defaults to False.

        Returns
        -------
        Manifest
        """
        if manifest_name not in self.manifest_file_names:
            raise ValueError(
                f"Manifest to load ({manifest_name}) is not one of the "
                "valid manifest names for this dataset. Valid names include:\n"
                f"{self.manifest_file_names}"
            )

        if use_static_project_dir:
            manifest_path = os.path.join(
                self._cache_dir, self.project_name, "manifests", manifest_name
            )
        else:
            manifest_path = os.path.join(self._cache_dir, manifest_name)

        with open(manifest_path, "r") as f:
            local_manifest = Manifest(
                cache_dir=self._cache_dir,
                json_input=f,
                use_static_project_dir=use_static_project_dir
            )

        return local_manifest

    def load_manifest(self, manifest_name: str):
        """
        Load a manifest from this dataset.

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        """
        self._manifest = self._load_manifest(manifest_name)
        self._manifest_name = manifest_name

    def _file_exists(self, file_attributes: CacheFileAttributes) -> bool:
        """
        Given a CacheFileAttributes describing a file, assess whether or
        not that file exists locally.

        Parameters
        ----------
        file_attributes: CacheFileAttributes
            Description of the file to look for

        Returns
        -------
        bool
            True if the file exists and is valid; False otherwise

        Raises
        -----
        RuntimeError
            If file_attributes.local_path exists but is not a file.
            It would be unclear how the cache should proceed in this case.
        """
        file_exists = False

        if file_attributes.local_path.exists():
            if not file_attributes.local_path.is_file():
                raise RuntimeError(f"{file_attributes.local_path}\n"
                                   "exists, but is not a file;\n"
                                   "unsure how to proceed")

            file_exists = True

        return file_exists

    def metadata_path(self, fname: str) -> dict:
        """
        Return the local path to a metadata file, and test for the
        file's existence

        Parameters
        ----------
        fname: str
            The name of the metadata file to be accessed

        Returns
        -------
        dict

            'path' will be a pathlib.Path pointing to the file's location

            'exists' will be a boolean indicating if the file
            exists in a valid state

            'file_attributes' is a CacheFileAttributes describing the file
            in more detail

        Raises
        ------
        RuntimeError
            If the file cannot be downloaded
        """
        file_attributes = self._manifest.metadata_file_attributes(fname)
        exists = self._file_exists(file_attributes)
        local_path = file_attributes.local_path
        output = {'local_path': local_path,
                  'exists': exists,
                  'file_attributes': file_attributes}

        return output

    def data_path(self, file_id) -> dict:
        """
        Return the local path to a data file, and test for the
        file's existence

        Parameters
        ----------
        file_id:
            The unique identifier of the file to be accessed

        Returns
        -------
        dict

            'local_path' will be a pathlib.Path pointing to the file's location

            'exists' will be a boolean indicating if the file
            exists in a valid state

            'file_attributes' is a CacheFileAttributes describing the file
            in more detail

        Raises
        ------
        RuntimeError
            If the file cannot be downloaded
        """
        file_attributes = self.get_file_attributes(file_id)
        exists = self._file_exists(file_attributes)
        local_path = file_attributes.local_path
        output = {'local_path': local_path,
                  'exists': exists,
                  'file_attributes': file_attributes}

        return output

    def get_file_attributes(self, file_id):
        """
        Retrieve file attributes for a given file_id from the meatadata.

        Parameters
        ----------
        file_id: str or int
            The unique identifier of the file to be accessed

        Returns
        -------
        CacheFileAttributes
        """
        return self._manifest.data_file_attributes(file_id)


class CloudCacheBase(BasicLocalCache):
    """
    A class to handle the downloading and accessing of data served from a cloud
    storage system

    Parameters
    ----------
    cache_dir: str or pathlib.Path
        Path to the directory where data will be stored on the local system

    project_name: str
        the name of the project this cache is supposed to access. This will
        be the root directory for all files stored in the bucket.

    ui_class_name: Optional[str]
        Name of the class users are actually using to manipulate this
        functionality (used to populate helpful error messages)
    """

    _bucket_name = None

    def __init__(self, cache_dir, project_name, ui_class_name=None):
        super().__init__(cache_dir=cache_dir, project_name=project_name,
                         ui_class_name=ui_class_name)

        # what latest_manifest was the last time an OutdatedManifestWarning
        # was emitted
        self._manifest_last_warned_on = None

        c_path = pathlib.Path(self._cache_dir)

        # self._manifest_last_used contains the name of the manifest
        # last loaded from this cache dir (if applicable)
        self._manifest_last_used = c_path / '_manifest_last_used.txt'

        # self._downloaded_data_path is where we will keep a JSONized
        # dict mapping paths to downloaded files to their file_hashes;
        # this will be used when determining if a downloaded file
        # can instead be a symlink
        self._downloaded_data_path = c_path / '_downloaded_data.json'

        # if the local manifest is missing but there are
        # data files in cache_dir, emit a warning
        # suggesting that the user run
        # self.construct_local_manifest
        if not self._downloaded_data_path.exists():
            file_list = c_path.glob('**/*')
            has_files = False
            for fname in file_list:
                if fname.is_file():
                    if 'json' not in fname.name:
                        has_files = True
                        break
            if has_files:
                msg = 'This cache directory appears to '
                msg += 'contain data files, but it has no '
                msg += 'record of what those files are. '
                msg += 'You might want to consider running\n\n'
                msg += f'{self.ui}.construct_local_manifest()\n\n'
                msg += 'to avoid needlessly downloading duplicates '
                msg += 'of data files that did not change between '
                msg += 'data releases. NOTE: running this method '
                msg += 'will require hashing every data file you '
                msg += 'have currently downloaded and could be '
                msg += 'very time consuming.\n\n'
                msg += 'To avoid this warning in the future, make '
                msg += 'sure that\n\n'
                msg += f'{str(self._downloaded_data_path.resolve())}\n\n'
                msg += 'is not deleted between instantiations of this '
                msg += 'cache'
                warnings.warn(msg, MissingLocalManifestWarning)

    def construct_local_manifest(self) -> None:
        """
        Construct the dict that maps between file_hash and
        absolute local path. Save it to self._downloaded_data_path
        """
        lookup = {}
        files_to_hash = set()
        c_dir = pathlib.Path(self._cache_dir)
        file_iterator = c_dir.glob('**/*')
        for file_name in file_iterator:
            if file_name.is_file():
                if 'json' not in file_name.name:
                    if file_name != self._manifest_last_used:
                        files_to_hash.add(file_name.resolve())

        with tqdm.tqdm(files_to_hash,
                       total=len(files_to_hash),
                       unit='(files hashed)') as pbar:

            for local_path in pbar:
                hsh = file_hash_from_path(local_path)
                lookup[str(local_path.absolute())] = hsh

        with open(self._downloaded_data_path, 'w') as out_file:
            out_file.write(json.dumps(lookup, indent=2, sort_keys=True))

    def _warn_of_outdated_manifest(self, manifest_name: str) -> None:
        """
        Warn that manifest_name is not the latest manifest available
        """
        if self._manifest_last_warned_on is not None:
            if self.latest_manifest_file == self._manifest_last_warned_on:
                return None

        self._manifest_last_warned_on = self.latest_manifest_file

        msg = '\n\n'
        msg += 'The manifest file you are loading is not the '
        msg += 'most up to date manifest file available for '
        msg += 'this dataset. The most up to data manifest file '
        msg += 'available for this dataset is \n\n'
        msg += f'{self.latest_manifest_file}\n\n'
        msg += 'To see the differences between these manifests,'
        msg += 'run\n\n'
        msg += f"{self.ui}.compare_manifests('{manifest_name}', "
        msg += f"'{self.latest_manifest_file}')\n\n"
        msg += "To see all of the manifest files currently downloaded "
        msg += "onto your local system, run\n\n"
        msg += "self.list_all_downloaded_manifests()\n\n"
        msg += "If you just want to load the latest manifest, run\n\n"
        msg += "self.load_latest_manifest()\n\n"
        warnings.warn(msg, OutdatedManifestWarning)
        return None

    @property
    def latest_downloaded_manifest_file(self) -> str:
        """parses downloaded available manifest files for semver string
        and returns the latest one
        self.manifest_file_names are assumed to be of the form
        '<anything>_v<semver_str>.json'

        Returns
        -------
        str
            the filename whose semver string is the latest one
        """
        file_list = self.list_all_downloaded_manifests()
        if len(file_list) == 0:
            return ''
        return self._find_latest_file(self.list_all_downloaded_manifests())

    def load_last_manifest(self):
        """
        If this Cache was used previously, load the last manifest
        used in this cache. If this cache has never been used, load
        the latest manifest.
        """
        if not self._manifest_last_used.exists():
            self.load_latest_manifest()
            return None

        with open(self._manifest_last_used, 'r') as in_file:
            to_load = in_file.read()

        latest = self.latest_manifest_file

        if to_load not in self.manifest_file_names:
            msg = 'The manifest version recorded as last used '
            msg += f'for this cache -- {to_load}-- '
            msg += 'is not a valid manifest for this dataset. '
            msg += f'Loading latest version -- {latest} -- '
            msg += 'instead.'
            warnings.warn(msg, UserWarning)
            self.load_latest_manifest()
            return None

        if latest != to_load:
            self._manifest_last_warned_on = self.latest_manifest_file
            msg = f"You are loading {to_load}. A more up to date "
            msg += f"version of the dataset -- {latest} -- exists "
            msg += "online. To see the changes between the two "
            msg += "versions of the dataset, run\n"
            msg += f"{self.ui}.compare_manifests('{to_load}',"
            msg += f" '{latest}')\n"
            msg += "To load another version of the dataset, run\n"
            msg += f"{self.ui}.load_manifest('{latest}')"
            warnings.warn(msg, OutdatedManifestWarning)
        self.load_manifest(to_load)
        return None

    def load_latest_manifest(self):
        latest_downloaded = self.latest_downloaded_manifest_file
        latest = self.latest_manifest_file
        if latest != latest_downloaded:
            if latest_downloaded != '':
                msg = f'You are loading\n{self.latest_manifest_file}\n'
                msg += 'which is newer than the most recent manifest '
                msg += 'file you have previously been working with\n'
                msg += f'{latest_downloaded}\n'
                msg += 'It is possible that some data files have changed '
                msg += 'between these two data releases, which will '
                msg += 'force you to re-download those data files '
                msg += '(currently downloaded files will not be overwritten).'
                msg += f' To continue using {latest_downloaded}, run\n'
                msg += f"{self.ui}.load_manifest('{latest_downloaded}')"
                warnings.warn(msg, OutdatedManifestWarning)
        self.load_manifest(self.latest_manifest_file)

    @abstractmethod
    def _download_manifest(self,
                           manifest_name: str):
        """
        Download a manifest from the dataset

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        """
        raise NotImplementedError()

    @abstractmethod
    def _download_file(self, file_attributes: CacheFileAttributes) -> bool:
        """
        Check if a file exists locally. If it does not, download it and
        return True. Return False otherwise.

        Parameters
        ----------
        file_attributes: CacheFileAttributes
            Describes the file to download

        Returns
        -------
        bool
            True if the file was downloaded; False otherwise

        Raises
        ------
        RuntimeError
            If the path to the directory where the file is to be saved
            points to something that is not a directory.

        RuntimeError
            If it is not able to successfully download the file after
            10 iterations
        """
        raise NotImplementedError()

    def load_manifest(self, manifest_name: str):
        """
        Load a manifest from this dataset.

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        """
        if manifest_name not in self.manifest_file_names:
            raise ValueError(
                f"Manifest to load ({manifest_name}) is not one of the "
                "valid manifest names for this dataset. Valid names include:\n"
                f"{self.manifest_file_names}"
            )

        if manifest_name != self.latest_manifest_file:
            self._warn_of_outdated_manifest(manifest_name)

        # If desired manifest does not exist, try to download it
        manifest_path = os.path.join(self._cache_dir, manifest_name)
        if not os.path.exists(manifest_path):
            self._download_manifest(manifest_name)

        self._manifest = self._load_manifest(manifest_name)

        # Keep track of the newly loaded manifest
        with open(self._manifest_last_used, 'w') as out_file:
            out_file.write(manifest_name)

        self._manifest_name = manifest_name

    def _update_list_of_downloads(self,
                                  file_attributes: CacheFileAttributes
                                  ) -> None:
        """
        Update the local file that keeps track of files that have actually
        been downloaded to reflect a newly downloaded file.

        Parameters
        ----------
        file_attributes: CacheFileAttributes

        Returns
        -------
        None
        """
        if not file_attributes.local_path.exists():
            # This file does not exist; there is nothing to do
            return None

        if self._downloaded_data_path.exists():
            with open(self._downloaded_data_path, 'rb') as in_file:
                downloaded_data = json.load(in_file)
        else:
            downloaded_data = {}

        abs_path = str(file_attributes.local_path.resolve())
        if abs_path in downloaded_data:
            if downloaded_data[abs_path] == file_attributes.file_hash:
                # this file has already been logged;
                # there is nothing to do
                return None

        downloaded_data[abs_path] = file_attributes.file_hash
        with open(self._downloaded_data_path, 'w') as out_file:
            out_file.write(json.dumps(downloaded_data,
                                      indent=2,
                                      sort_keys=True))
        return None

    def _check_for_identical_copy(self,
                                  file_attributes: CacheFileAttributes
                                  ) -> bool:
        """
        Check the manifest of files that have been locally downloaded to
        see if a file with an identical hash to the requested file has already
        been downloaded. If it has, create a symlink to the downloaded file
        at the requested file's localpath, update the manifest of downloaded
        files, and return True.

        Else return False

        Parameters
        ----------
        file_attributes: CacheFileAttributes
            The file we are considering downloading

        Returns
        -------
        bool
        """
        if not self._downloaded_data_path.exists():
            return False

        with open(self._downloaded_data_path, 'rb') as in_file:
            available_files = json.load(in_file)

        matched_path = None
        for abs_path in available_files:
            if available_files[abs_path] == file_attributes.file_hash:
                matched_path = pathlib.Path(abs_path)

                # check that the file still exists,
                # in case someone accidentally deleted
                # the file at the root of a symlink
                if matched_path.is_file():
                    break
                else:
                    matched_path = None

        if matched_path is None:
            return False

        local_parent = file_attributes.local_path.parent.resolve()
        if not local_parent.exists():
            os.makedirs(local_parent)

        file_attributes.local_path.symlink_to(matched_path.resolve())
        return True

    def _file_exists(self, file_attributes: CacheFileAttributes) -> bool:
        """
        Given a CacheFileAttributes describing a file, assess whether or
        not that file exists locally and is valid (i.e. has the expected
        file hash)

        Parameters
        ----------
        file_attributes: CacheFileAttributes
            Description of the file to look for

        Returns
        -------
        bool
            True if the file exists and is valid; False otherwise

        Raises
        -----
        RuntimeError
            If file_attributes.local_path exists but is not a file.
            It would be unclear how the cache should proceed in this case.
        """
        file_exists = False

        if file_attributes.local_path.exists():
            if not file_attributes.local_path.is_file():
                raise RuntimeError(f"{file_attributes.local_path}\n"
                                   "exists, but is not a file;\n"
                                   "unsure how to proceed")

            file_exists = True

        if not file_exists:
            file_exists = self._check_for_identical_copy(file_attributes)

        return file_exists

    def download_data(self, file_id) -> pathlib.Path:
        """
        Return the local path to a data file, downloading the file
        if necessary

        Parameters
        ----------
        file_id:
            The unique identifier of the file to be accessed

        Returns
        -------
        pathlib.Path
            The path indicating where the file is stored on the
            local system

        Raises
        ------
        RuntimeError
            If the file cannot be downloaded
        """
        super_attributes = self.data_path(file_id)
        file_attributes = super_attributes['file_attributes']
        was_downloaded = self._download_file(file_attributes)
        if was_downloaded:
            self._update_list_of_downloads(file_attributes)
        return file_attributes.local_path

    def download_metadata(self, fname: str) -> pathlib.Path:
        """
        Return the local path to a metadata file, downloading the
        file if necessary

        Parameters
        ----------
        fname: str
            The name of the metadata file to be accessed

        Returns
        -------
        pathlib.Path
            The path indicating where the file is stored on the
            local system

        Raises
        ------
        RuntimeError
            If the file cannot be downloaded
        """
        super_attributes = self.metadata_path(fname)
        file_attributes = super_attributes['file_attributes']
        was_downloaded = self._download_file(file_attributes)
        if was_downloaded:
            self._update_list_of_downloads(file_attributes)
        return file_attributes.local_path

    def get_metadata(self, fname: str) -> pd.DataFrame:
        """
        Return a pandas DataFrame of metadata

        Parameters
        ----------
        fname: str
            The name of the metadata file to load

        Returns
        -------
        pd.DataFrame

        Notes
        -----
        This method will check to see if the specified metadata file exists
        locally. If it does not, the method will download the file. Use
        self.metadata_path() to find where the file is stored
        """
        local_path = self.download_metadata(fname)
        return pd.read_csv(local_path)

    def _detect_changes(self,
                        filename_to_hash: dict) -> List[Tuple[str, str]]:
        """
        Assemble list of changes between two manifests

        Parameters
        ----------
        filename_to_hash: dict
            filename_to_hash[0] is a dict mapping file names to file hashes
            for manifest 0

            filename_to_hash[1] is a dict mapping file names to file hashes
            for manifest 1

        Returns
        -------
        List[Tuple[str, str]]
            List of changes between manifest 0 and manifest 1.

        Notes
        -----
        Changes are tuples of the form
        (fname, string describing how fname changed)

        e.g.

        ('data/f1.txt', 'data/f1.txt renamed data/f5.txt')
        ('data/f2.txt', 'data/f2.txt deleted')
        ('data/f3.txt', 'data/f3.txt created')
        ('data/f4.txt', 'data/f4.txt changed')
        """
        output = []
        n0 = set(filename_to_hash[0].keys())
        n1 = set(filename_to_hash[1].keys())
        all_file_names = n0.union(n1)

        hash_to_filename: dict = dict()
        for v in (0, 1):
            hash_to_filename[v] = {}
            for fname in filename_to_hash[v]:
                hash_to_filename[v][filename_to_hash[v][fname]] = fname

        for fname in all_file_names:
            delta = None
            if fname in filename_to_hash[0] and fname in filename_to_hash[1]:
                h0 = filename_to_hash[0][fname]
                h1 = filename_to_hash[1][fname]
                if h0 != h1:
                    delta = f'{fname} changed'
            elif fname in filename_to_hash[0]:
                h0 = filename_to_hash[0][fname]
                if h0 in hash_to_filename[1]:
                    f1 = hash_to_filename[1][h0]
                    delta = f'{fname} renamed {f1}'
                else:
                    delta = f'{fname} deleted'
            elif fname in filename_to_hash[1]:
                h1 = filename_to_hash[1][fname]
                if h1 not in hash_to_filename[0]:
                    delta = f'{fname} created'
            else:
                raise RuntimeError("should never reach this line")

            if delta is not None:
                output.append((fname, delta))

        return output

    def summarize_comparison(self,
                             manifest_0_name: str,
                             manifest_1_name: str
                             ) -> Dict[str, List[Tuple[str, str]]]:
        """
        Compare two manifests from this dataset. Return a dict
        containing the list of metadata and data files that changed
        between them

        Note: this assumes that manifest_0 predates manifest_1 (i.e.
        changes are listed relative to manifest_0)

        Parameters
        ----------
        manifest_0_name: str

        manifest_1_name: str

        Returns
        -------
        result: Dict[List[Tuple[str, str]]]
            result['data_changes'] lists changes to data files
            result['metadata_changes'] lists changes to metadata files

        Notes
        -----
        Changes are tuples of the form
        (fname, string describing how fname changed)

        e.g.

        ('data/f1.txt', 'data/f1.txt renamed data/f5.txt')
        ('data/f2.txt', 'data/f2.txt deleted')
        ('data/f3.txt', 'data/f3.txt created')
        ('data/f4.txt', 'data/f4.txt changed')
        """
        for manifest_name in [manifest_0_name, manifest_1_name]:
            manifest_path = os.path.join(self._cache_dir, manifest_name)
            if not os.path.exists(manifest_path):
                self._download_manifest(manifest_name)

        man0 = self._load_manifest(manifest_0_name)
        man1 = self._load_manifest(manifest_1_name)

        result: dict = dict()
        for (result_key,
             file_id_list,
             attr_lookup) in zip(('metadata_changes', 'data_changes'),
                                 ((man0.metadata_file_names,
                                   man1.metadata_file_names),
                                  (man0.file_id_values,
                                   man1.file_id_values)),
                                 ((man0.metadata_file_attributes,
                                   man1.metadata_file_attributes),
                                  (man0.data_file_attributes,
                                   man1.data_file_attributes))):

            filename_to_hash: dict = dict()
            for version in (0, 1):
                filename_to_hash[version] = {}
                for file_id in file_id_list[version]:
                    obj = attr_lookup[version](file_id)
                    file_name = relative_path_from_url(obj.url)
                    file_name = '/'.join(file_name.split('/')[1:])
                    filename_to_hash[version][file_name] = obj.file_hash
            changes = self._detect_changes(filename_to_hash)
            result[result_key] = changes
        return result

    def compare_manifests(self,
                          manifest_0_name: str,
                          manifest_1_name: str
                          ) -> str:
        """
        Compare two manifests from this dataset. Return a dict
        containing the list of metadata and data files that changed
        between them

        Note: this assumes that manifest_0 predates manifest_1

        Parameters
        ----------
        manifest_0_name: str

        manifest_1_name: str

        Returns
        -------
        str
            A string summarizing all of the changes going from
            manifest_0 to manifest_1
        """

        changes = self.summarize_comparison(manifest_0_name,
                                            manifest_1_name)
        if len(changes['data_changes']) == 0:
            if len(changes['metadata_changes']) == 0:
                return "The two manifests are equivalent"

        data_change_dict = {}
        for delta in changes['data_changes']:
            data_change_dict[delta[0]] = delta[1]
        metadata_change_dict = {}
        for delta in changes['metadata_changes']:
            metadata_change_dict[delta[0]] = delta[1]

        msg = 'Changes going from\n'
        msg += f'{manifest_0_name}\n'
        msg += 'to\n'
        msg += f'{manifest_1_name}\n\n'

        m_keys = list(metadata_change_dict.keys())
        m_keys.sort()
        for m in m_keys:
            msg += f'{metadata_change_dict[m]}\n'
        d_keys = list(data_change_dict.keys())
        d_keys.sort()
        for d in d_keys:
            msg += f'{data_change_dict[d]}\n'
        return msg


class S3CloudCache(CloudCacheBase):
    """
    A class to handle the downloading and accessing of data served from
    an S3-based storage system

    Parameters
    ----------
    cache_dir: str or pathlib.Path
        Path to the directory where data will be stored on the local system

    bucket_name: str
        for example, if bucket URI is 's3://mybucket' this value should be
        'mybucket'

    project_name: str
        the name of the project this cache is supposed to access. This will
        be the root directory for all files stored in the bucket.

    ui_class_name: Optional[str]
        Name of the class users are actually using to maniuplate this
        functionality (used to populate helpful error messages)
    """

    def __init__(self, cache_dir, bucket_name, project_name,
                 ui_class_name=None):
        self._manifest = None
        self._bucket_name = bucket_name

        super().__init__(cache_dir=cache_dir, project_name=project_name,
                         ui_class_name=ui_class_name)

    _s3_client = None

    @property
    def bucket_name(self) -> str:
        return self._bucket_name

    @property
    def s3_client(self):
        if self._s3_client is None:
            s3_config = Config(signature_version=UNSIGNED)
            self._s3_client = boto3.client('s3',
                                           config=s3_config)
        return self._s3_client

    def _list_all_manifests(self) -> list:
        """
        Return a list of all of the file names of the manifests associated
        with this dataset
        """
        paginator = self.s3_client.get_paginator('list_objects_v2')
        subset_iterator = paginator.paginate(
            Bucket=self._bucket_name,
            Prefix=self.manifest_prefix
        )

        output = []
        for subset in subset_iterator:
            if 'Contents' in subset:
                for obj in subset['Contents']:
                    output.append(pathlib.Path(obj['Key']).name)

        output.sort()
        return output

    def _download_manifest(self,
                           manifest_name: str):
        """
        Download a manifest from the dataset

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        """

        manifest_key = self.manifest_prefix + manifest_name
        response = self.s3_client.get_object(Bucket=self._bucket_name,
                                             Key=manifest_key)

        filepath = os.path.join(self._cache_dir, manifest_name)

        with open(filepath, 'wb') as f:
            for chunk in response['Body'].iter_chunks():
                f.write(chunk)

    def _download_file(self, file_attributes: CacheFileAttributes) -> bool:
        """
        Check if a file exists locally. If it does not, download it
        and return True. Return False otherwise.

        Parameters
        ----------
        file_attributes: CacheFileAttributes
            Describes the file to download

        Returns
        -------
        bool
            True if the file was downloaded; False otherwise

        Raises
        ------
        RuntimeError
            If the path to the directory where the file is to be saved
            points to something that is not a directory.

        RuntimeError
            If it is not able to successfully download the file after
            10 iterations
        """
        was_downloaded = False

        local_path = file_attributes.local_path

        local_dir = pathlib.Path(safe_system_path(str(local_path.parents[0])))

        # make sure Windows references to Allen Institute
        # local networked file system get handled correctly
        local_path = pathlib.Path(safe_system_path(str(local_path)))

        # using os here rather than pathlib because safe_system_path
        # returns a str
        os.makedirs(local_dir, exist_ok=True)
        if not os.path.isdir(local_dir):
            raise RuntimeError(f"{local_dir}\n"
                               "is not a directory")

        bucket_name = bucket_name_from_url(file_attributes.url)
        obj_key = relative_path_from_url(file_attributes.url)

        n_iter = 0
        max_iter = 10  # maximum number of times to try download

        version_id = file_attributes.version_id

        pbar = None
        if not self._file_exists(file_attributes):
            response = self.s3_client.list_object_versions(Bucket=bucket_name,
                                                           Prefix=str(obj_key))
            object_info = [i for i in response["Versions"]
                           if i["VersionId"] == version_id][0]
            pbar = tqdm.tqdm(desc=object_info["Key"].split("/")[-1],
                             total=object_info["Size"],
                             unit_scale=True,
                             unit_divisor=1000.,
                             unit="MB")

        while not self._file_exists(file_attributes):
            was_downloaded = True
            response = self.s3_client.get_object(Bucket=bucket_name,
                                                 Key=str(obj_key),
                                                 VersionId=version_id)

            if 'Body' in response:
                with open(local_path, 'wb') as out_file:
                    for chunk in response['Body'].iter_chunks():
                        out_file.write(chunk)
                        pbar.update(len(chunk))

            # Verify the hash of the downloaded file
            full_path = file_attributes.local_path.resolve()
            test_checksum = file_hash_from_path(full_path)
            if test_checksum != file_attributes.file_hash:
                file_attributes.local_path.exists()
                file_attributes.local_path.unlink()

            n_iter += 1
            if n_iter > max_iter:
                pbar.close()
                raise RuntimeError("Could not download\n"
                                   f"{file_attributes}\n"
                                   "In {max_iter} iterations")
        if pbar is not None:
            pbar.close()

        return was_downloaded


class LocalCache(CloudCacheBase):
    """A class to handle accessing of data that has already been downloaded
    locally. Supports multiple manifest versions from a given dataset.

    Parameters
    ----------
    cache_dir: str or pathlib.Path
        Path to the directory where data will be stored on the local system

    project_name: str
        the name of the project this cache is supposed to access. This will
        be the root directory for all files stored in the bucket.

    ui_class_name: Optional[str]
        Name of the class users are actually using to maniuplate this
        functionality (used to populate helpful error messages)
    """
    def __init__(self, cache_dir, project_name, ui_class_name=None):
        super().__init__(cache_dir=cache_dir, project_name=project_name,
                         ui_class_name=ui_class_name)

    def _list_all_manifests(self) -> list:
        return self.list_all_downloaded_manifests()

    def _download_manifest(self, manifest_name: str):
        raise NotImplementedError()

    def _download_file(self, file_attributes: CacheFileAttributes) -> bool:
        raise NotImplementedError()


class StaticLocalCache(BasicLocalCache):
    """A class to handle accessing data that has already been downloaded
    locally and whose directory structure and/or contained files are not
    expected to be changed in any way. Does NOT support multiple manifest
    versions for a given dataset.

    Example intended use case:
        Calling
        VisualBehaviorOphysProjectCache.from_local_cache(use_static_cache=True)
        where the cache directory is a mounted S3 public bucket.

    Parameters
    ----------
    cache_dir: str or pathlib.Path
        Path to the directory where data will be stored on the local system

    project_name: str
        the name of the project this cache is supposed to access. This will
        be the root directory for all files stored in the bucket.

    ui_class_name: Optional[str]
        Name of the class users are actually using to maniuplate this
        functionality (used to populate helpful error messages)
    """

    def __init__(self, cache_dir, project_name, ui_class_name=None):
        super().__init__(cache_dir=cache_dir, project_name=project_name,
                         ui_class_name=ui_class_name)

    def _list_all_manifests(self) -> list:
        """
        Return a list of all of the file names of the manifests associated
        with this dataset. For the StaticLocalCache only return only the
        latest manifest.
        """
        manifest_dir = os.path.join(
            self._cache_dir, self.project_name, "manifests"
        )
        if not os.path.exists(manifest_dir):
            raise RuntimeError(
                f"Expected the provided cache_dir ({self._cache_dir})"
                "to have the following subfolders but it did not: "
                f"{self.project_name}/manifests"
            )

        output = [x for x in os.listdir(manifest_dir)
                  if re.fullmatch(".*_manifest_v.*.json", x)]

        return [self._find_latest_file(output)]

    def list_all_downloaded_manifests(self) -> list:
        """
        Return a list of all of the manifest files for this dataset.
        For the StaticLocalCache, this will only be the latest manifest.
        """
        return self._list_all_manifests()

    def load_last_manifest(self):
        """For the StaticLocalCache always load the latest manifest."""
        self.load_manifest(self.latest_manifest_file)

    def load_manifest(self, manifest_name: str):
        """
        Load a manifest from this dataset.

        Parameters
        ----------
        manifest_name: str
            The name of the manifest to load. Must be an element in
            self.manifest_file_names
        """
        self._manifest = self._load_manifest(
            manifest_name,
            use_static_project_dir=True
        )
        self._manifest_name = manifest_name

    def compare_manifests(self, manifest_0_name: str, manifest_1_name: str):
        raise RuntimeError(
            "The ability to load many manifest versions and use the "
            "`compare_manifests()` method is not available for the "
            "StaticLocalCache class!"
        )