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
[docs]
class OutdatedManifestWarning(UserWarning):
pass
[docs]
class MissingLocalManifestWarning(UserWarning):
pass
[docs]
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()
[docs]
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
[docs]
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
[docs]
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
[docs]
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)
[docs]
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)
[docs]
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())
[docs]
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
[docs]
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()
[docs]
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
[docs]
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 _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
[docs]
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
[docs]
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
[docs]
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
[docs]
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()
[docs]
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)]
[docs]
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()
[docs]
def load_last_manifest(self):
"""For the StaticLocalCache always load the latest manifest."""
self.load_manifest(self.latest_manifest_file)
[docs]
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
[docs]
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!"
)