Python源码示例：tarfile.extractall()
示例1
def extract_tarfile(tar_location, tar_flag, extract_location, name_of_file):
"""Attempts to extract a tar file (tgz).
If the file is not found, or the extraction of the contents failed, the
exception will be caught and the function will return False. If successful,
the tar file will be removed.
Args:
tar_location: A string of the current location of the tgz file
tar_flag: A string indicating the mode to open the tar file.
tarfile.extractall will generate an error if a flag with permissions
other than read is passed.
extract_location: A string of the path the file will be extracted to.
name_of_file: A string with the name of the file, for printing purposes.
Returns:
Boolean: Whether or not the tar extraction succeeded.
"""
try:
with tarfile.open(tar_location, tar_flag) as tar:
tar.extractall(path=extract_location, members=tar)
os.remove(tar_location)
print "\t" + name_of_file + " successfully extracted."
return True
except tarfile.ExtractError:
return False
示例2
def generator(self, data_dir, tmp_dir, datasets,
eos_list=None, start_from=0, how_many=0):
del eos_list
i = 0
for url, subdir in datasets:
filename = os.path.basename(url)
compressed_file = generator_utils.maybe_download(tmp_dir, filename, url)
read_type = "r:gz" if filename.endswith("tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow
members = []
for f in corpus_tar:
if not os.path.isfile(os.path.join(tmp_dir, f.name)):
members.append(f)
corpus_tar.extractall(tmp_dir, members=members)
data_dir = os.path.join(tmp_dir, "LibriSpeech", subdir)
data_files = _collect_data(data_dir, "flac", "txt")
data_pairs = data_files.values()
encoders = self.feature_encoders(None)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for utt_id, media_file, text_data in sorted(data_pairs)[start_from:]:
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
spk_id, unused_book_id, _ = utt_id.split("-")
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": [spk_id],
}
示例3
def generator(self, data_dir, tmp_dir, datasets,
eos_list=None, start_from=0, how_many=0):
del eos_list
i = 0
for url, subdir in datasets:
filename = os.path.basename(url)
compressed_file = generator_utils.maybe_download(tmp_dir, filename, url)
read_type = "r:gz" if filename.endswith("tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow
members = []
for f in corpus_tar:
if not os.path.isfile(os.path.join(tmp_dir, f.name)):
members.append(f)
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "LibriSpeech", subdir)
data_files = _collect_data(raw_data_dir, "flac", "txt")
data_pairs = data_files.values()
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for utt_id, media_file, text_data in sorted(data_pairs)[start_from:]:
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
spk_id, unused_book_id, _ = utt_id.split("-")
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": [spk_id],
}
示例4
def generator(self, data_dir, tmp_dir, datasets,
eos_list=None, start_from=0, how_many=0):
del eos_list
i = 0
for url, subdir in datasets:
filename = os.path.basename(url)
compressed_file = generator_utils.maybe_download(tmp_dir, filename, url)
read_type = "r:gz" if filename.endswith("tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow
members = []
for f in corpus_tar:
if not os.path.isfile(os.path.join(tmp_dir, f.name)):
members.append(f)
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "LibriSpeech", subdir)
data_files = _collect_data(raw_data_dir, "flac", "txt")
data_pairs = data_files.values()
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for utt_id, media_file, text_data in sorted(data_pairs)[start_from:]:
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
spk_id, unused_book_id, _ = utt_id.split("-")
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": [spk_id],
}
示例5
def generator(self, data_dir, tmp_dir, datasets,
eos_list=None, start_from=0, how_many=0):
del eos_list
i = 0
for url, subdir in datasets:
filename = os.path.basename(url)
compressed_file = generator_utils.maybe_download(tmp_dir, filename, url)
read_type = "r:gz" if filename.endswith("tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow
members = []
for f in corpus_tar:
if not os.path.isfile(os.path.join(tmp_dir, f.name)):
members.append(f)
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "LibriSpeech", subdir)
data_files = _collect_data(raw_data_dir, "flac", "txt")
data_pairs = data_files.values()
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for utt_id, media_file, text_data in sorted(data_pairs)[start_from:]:
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
spk_id, unused_book_id, _ = utt_id.split("-")
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": [spk_id],
}
示例6
def generator(self, data_dir, tmp_dir, datasets,
eos_list=None, start_from=0, how_many=0):
del eos_list
i = 0
for url, subdir in datasets:
filename = os.path.basename(url)
compressed_file = generator_utils.maybe_download(tmp_dir, filename, url)
read_type = "r:gz" if filename.endswith("tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow
members = []
for f in corpus_tar:
if not os.path.isfile(os.path.join(tmp_dir, f.name)):
members.append(f)
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "LibriSpeech", subdir)
data_files = _collect_data(raw_data_dir, "flac", "txt")
data_pairs = data_files.values()
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for utt_id, media_file, text_data in sorted(data_pairs)[start_from:]:
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
spk_id, unused_book_id, _ = utt_id.split("-")
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": [spk_id],
}
示例7
def extract_zipfile(zip_location, zip_flag, extract_location, name_of_file):
"""Attempts to extract a zip file (zip).
If the file is not found, or the extraction of the contents failed, the
exception will be caught and the function will return False. If successful,
the zip file will be removed.
Args:
zip_location: A string of the current location of the zip file
zip_flag: A string indicating the mode to open the zip file.
zipfile.extractall will generate an error if a flag with permissions
other than read is passed.
extract_location: A string of the path the file will be extracted to.
name_of_file: A string with the name of the file, for printing purposes.
Returns:
Boolean: Whether or not the zip extraction succeeded.
"""
try:
with zipfile.ZipFile(zip_location, zip_flag) as zf:
zf.extractall(extract_location)
os.remove(zip_location)
print "\t" + name_of_file + " successfully extracted."
return True
except zipfile.BadZipfile:
sys.stderr.write("\t" + name_of_file + " failed to extract.\n")
return False
示例8
def generator(self,
data_dir,
tmp_dir,
datasets,
eos_list=None,
start_from=0,
how_many=0):
del eos_list
i = 0
filename = os.path.basename(_COMMONVOICE_URL)
compressed_file = generator_utils.maybe_download(tmp_dir, filename,
_COMMONVOICE_URL)
read_type = "r:gz" if filename.endswith(".tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow. For security, check that all
# paths are relative.
members = [
f for f in corpus_tar if _is_relative(tmp_dir, f.name) and
not _file_exists(tmp_dir, f.name)
]
corpus_tar.extractall(tmp_dir, members=members)
data_dir = os.path.join(tmp_dir, "cv_corpus_v1")
data_tuples = _collect_data(data_dir)
encoders = self.feature_encoders(None)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for dataset in datasets:
data_tuples = (tup for tup in data_tuples if tup[0].startswith(dataset))
for utt_id, media_file, text_data in tqdm.tqdm(
sorted(data_tuples)[start_from:]):
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": ["unknown"],
}
示例9
def generator(self,
data_dir,
tmp_dir,
datasets,
eos_list=None,
start_from=0,
how_many=0):
del eos_list
i = 0
filename = os.path.basename(_COMMONVOICE_URL)
compressed_file = generator_utils.maybe_download(tmp_dir, filename,
_COMMONVOICE_URL)
read_type = "r:gz" if filename.endswith(".tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow. For security, check that all
# paths are relative.
members = [
f for f in corpus_tar if _is_relative(tmp_dir, f.name) and
not _file_exists(tmp_dir, f.name)
]
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "cv_corpus_v1")
data_tuples = _collect_data(raw_data_dir)
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for dataset in datasets:
data_tuples = (tup for tup in data_tuples if tup[0].startswith(dataset))
for utt_id, media_file, text_data in tqdm.tqdm(
sorted(data_tuples)[start_from:]):
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": ["unknown"],
}
示例10
def generator(self,
data_dir,
tmp_dir,
datasets,
eos_list=None,
start_from=0,
how_many=0):
del eos_list
i = 0
filename = os.path.basename(_COMMONVOICE_URL)
compressed_file = generator_utils.maybe_download(tmp_dir, filename,
_COMMONVOICE_URL)
read_type = "r:gz" if filename.endswith(".tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow. For security, check that all
# paths are relative.
members = [
f for f in corpus_tar if _is_relative(tmp_dir, f.name) and
not _file_exists(tmp_dir, f.name)
]
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "cv_corpus_v1")
data_tuples = _collect_data(raw_data_dir)
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for dataset in datasets:
data_tuples = (tup for tup in data_tuples if tup[0].startswith(dataset))
for utt_id, media_file, text_data in tqdm.tqdm(
sorted(data_tuples)[start_from:]):
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": ["unknown"],
}
示例11
def _analyze_tarfile_for_import(tarfile, project, schema, tmpdir):
def read_sp_manifest_file(path):
# Must use forward slashes, not os.path.sep.
fn_manifest = _tarfile_path_join(path, project.Job.FN_MANIFEST)
try:
with closing(tarfile.extractfile(fn_manifest)) as file:
if sys.version_info < (3, 6):
return json.loads(file.read().decode())
else:
return json.loads(file.read())
except KeyError:
pass
if schema is None:
schema_function = read_sp_manifest_file
elif callable(schema):
schema_function = _with_consistency_check(schema, read_sp_manifest_file)
elif isinstance(schema, str):
schema_function = _with_consistency_check(
_make_path_based_schema_function(schema), read_sp_manifest_file)
else:
raise TypeError("The schema variable must be None, callable, or a string.")
mappings = dict()
skip_subdirs = set()
dirs = [member.name for member in tarfile.getmembers() if member.isdir()]
for name in sorted(dirs):
if os.path.dirname(name) in skip_subdirs: # skip all sub-dirs of identified dirs
skip_subdirs.add(name)
continue
sp = schema_function(name)
if sp is not None:
job = project.open_job(sp)
if os.path.exists(job.workspace()):
raise DestinationExistsError(job)
mappings[name] = job
skip_subdirs.add(name)
# Check uniqueness
if len(set(mappings.values())) != len(mappings):
raise StatepointParsingError(
"The jobs identified with the given schema function are not unique!")
tarfile.extractall(path=tmpdir)
for path, job in mappings.items():
src = os.path.join(tmpdir, path)
assert os.path.isdir(tmpdir)
assert os.path.isdir(src)
yield src, _CopyFromTarFileExecutor(src, job)
示例12
def generator(self,
data_dir,
tmp_dir,
datasets,
eos_list=None,
start_from=0,
how_many=0):
del eos_list
i = 0
filename = os.path.basename(_COMMONVOICE_URL)
compressed_file = generator_utils.maybe_download(tmp_dir, filename,
_COMMONVOICE_URL)
read_type = "r:gz" if filename.endswith(".tgz") else "r"
with tarfile.open(compressed_file, read_type) as corpus_tar:
# Create a subset of files that don't already exist.
# tarfile.extractall errors when encountering an existing file
# and tarfile.extract is extremely slow. For security, check that all
# paths are relative.
members = [
f for f in corpus_tar if _is_relative(tmp_dir, f.name) and
not _file_exists(tmp_dir, f.name)
]
corpus_tar.extractall(tmp_dir, members=members)
raw_data_dir = os.path.join(tmp_dir, "cv_corpus_v1")
data_tuples = _collect_data(raw_data_dir)
encoders = self.feature_encoders(data_dir)
audio_encoder = encoders["waveforms"]
text_encoder = encoders["targets"]
for dataset in datasets:
data_tuples = (tup for tup in data_tuples if tup[0].startswith(dataset))
for utt_id, media_file, text_data in tqdm.tqdm(
sorted(data_tuples)[start_from:]):
if how_many > 0 and i == how_many:
return
i += 1
wav_data = audio_encoder.encode(media_file)
yield {
"waveforms": wav_data,
"waveform_lens": [len(wav_data)],
"targets": text_encoder.encode(text_data),
"raw_transcript": [text_data],
"utt_id": [utt_id],
"spk_id": ["unknown"],
}
示例13
def download_nidm_pain(data_dir=None, overwrite=False, verbose=1):
"""
Download NIDM Results for 21 pain studies from NeuroVault for tests.
Parameters
----------
data_dir : :obj:`str`, optional
Location in which to place the studies. Default is None, which uses the
package's default path for downloaded data.
overwrite : :obj:`bool`, optional
Whether to overwrite existing files or not. Default is False.
verbose : :obj:`int`, optional
Default is 1.
Returns
-------
data_dir : :obj:`str`
Updated data directory pointing to dataset files.
"""
url = 'https://neurovault.org/collections/1425/download'
dataset_name = 'nidm_21pain'
data_dir = _get_dataset_dir(dataset_name, data_dir=data_dir, verbose=verbose)
desc_file = op.join(data_dir, 'description.txt')
if op.isfile(desc_file) and overwrite is False:
return data_dir
# Download
fname = op.join(data_dir, url.split('/')[-1])
_download_zipped_file(url, filename=fname)
# Unzip
with zipfile.ZipFile(fname, 'r') as zip_ref:
zip_ref.extractall(data_dir)
collection_folders = [f for f in glob(op.join(data_dir, '*')) if '.nidm' not in f]
collection_folders = [f for f in collection_folders if op.isdir(f)]
if len(collection_folders) > 1:
raise Exception('More than one folder found: '
'{0}'.format(', '.join(collection_folders)))
else:
folder = collection_folders[0]
zip_files = glob(op.join(folder, '*.zip'))
for zf in zip_files:
fn = op.splitext(op.basename(zf))[0]
with zipfile.ZipFile(zf, 'r') as zip_ref:
zip_ref.extractall(op.join(data_dir, fn))
os.remove(fname)
shutil.rmtree(folder)
with open(desc_file, 'w') as fo:
fo.write('21 pain studies in NIDM-results packs.')
return data_dir
示例14
def download_mallet(data_dir=None, overwrite=False, verbose=1):
"""
Download the MALLET toolbox for LDA topic modeling.
Parameters
----------
data_dir : :obj:`str`, optional
Location in which to place MALLET. Default is None, which uses the
package's default path for downloaded data.
overwrite : :obj:`bool`, optional
Whether to overwrite existing files or not. Default is False.
verbose : :obj:`int`, optional
Default is 1.
Returns
-------
data_dir : :obj:`str`
Updated data directory pointing to MALLET files.
"""
url = 'http://mallet.cs.umass.edu/dist/mallet-2.0.7.tar.gz'
temp_dataset_name = 'mallet__temp'
temp_data_dir = _get_dataset_dir(temp_dataset_name, data_dir=data_dir, verbose=verbose)
dataset_name = 'mallet'
data_dir = temp_data_dir.replace(temp_dataset_name, dataset_name)
desc_file = op.join(data_dir, 'description.txt')
if op.isfile(desc_file) and overwrite is False:
shutil.rmtree(temp_data_dir)
return data_dir
mallet_file = op.join(temp_data_dir, op.basename(url))
_download_zipped_file(url, mallet_file)
with tarfile.open(mallet_file) as tf:
tf.extractall(path=temp_data_dir)
os.rename(op.join(temp_data_dir, 'mallet-2.0.7'), data_dir)
os.remove(mallet_file)
shutil.rmtree(temp_data_dir)
with open(desc_file, 'w') as fo:
fo.write('The MALLET toolbox for latent Dirichlet allocation.')
if verbose > 0:
print('\nDataset moved to {}\n'.format(data_dir))
return data_dir
示例15
def download_peaks2maps_model(data_dir=None, overwrite=False, verbose=1):
"""
Download the trained Peaks2Maps model from OHBM 2018.
"""
url = "https://zenodo.org/record/1257721/files/ohbm2018_model.tar.xz?download=1"
temp_dataset_name = 'peaks2maps_model_ohbm2018__temp'
temp_data_dir = _get_dataset_dir(temp_dataset_name, data_dir=data_dir, verbose=verbose)
dataset_name = 'peaks2maps_model_ohbm2018'
data_dir = temp_data_dir.replace(temp_dataset_name, dataset_name)
desc_file = op.join(data_dir, 'description.txt')
if op.isfile(desc_file) and overwrite is False:
shutil.rmtree(temp_data_dir)
return data_dir
LGR.info('Downloading the model (this is a one-off operation)...')
# Streaming, so we can iterate over the response.
r = requests.get(url, stream=True)
f = BytesIO()
# Total size in bytes.
total_size = int(r.headers.get('content-length', 0))
block_size = 1024 * 1024
wrote = 0
for data in tqdm(r.iter_content(block_size), total=math.ceil(total_size // block_size),
unit='MB', unit_scale=True):
wrote = wrote + len(data)
f.write(data)
if total_size != 0 and wrote != total_size:
raise Exception("Download interrupted")
f.seek(0)
LGR.info('Uncompressing the model to {}...'.format(temp_data_dir))
tarfile = TarFile(fileobj=LZMAFile(f), mode="r")
tarfile.extractall(temp_data_dir)
os.rename(op.join(temp_data_dir, 'ohbm2018_model'), data_dir)
shutil.rmtree(temp_data_dir)
with open(desc_file, 'w') as fo:
fo.write('The trained Peaks2Maps model from OHBM 2018.')
if verbose > 0:
print('\nDataset moved to {}\n'.format(data_dir))
return data_dir
