# Copyright 2018 The TensorFlow Authors. All Rights Reserved.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

# ==============================================================================

# pylint: disable=g-import-not-at-top

"""Utilities for file download and caching."""

from __future__ import absolute_import

from __future__ import division

from __future__ import print_function

from abc import abstractmethod

from contextlib import closing

import gc

import hashlib

import multiprocessing

from multiprocessing.pool import ThreadPool

import os

import random

import shutil

import signal

import sys

import tarfile

import threading

import time

import weakref

import zipfile

import numpy as np

import six

from six.moves.urllib.error import HTTPError

from six.moves.urllib.error import URLError

from six.moves.urllib.request import urlopen

from tensorflow.python.keras.utils.generic_utils import Progbar

from tensorflow.python.util import tf_inspect

from tensorflow.python.util.tf_export import keras_export

try:

import queue

except ImportError:

import Queue as queue

if sys.version_info[0] == 2:

def urlretrieve(url, filename, reporthook=None, data=None):

fd.write(chunk)

else:

from six.moves.urllib.request import urlretrieve

def is_generator_or_sequence(x):

"""Check if `x` is a Keras generator type."""

return tf_inspect.isgenerator(x) or isinstance(x, Sequence)

def _extract_archive(file_path, path='.', archive_format='auto'):

return False

@keras_export('keras.utils.get_file')

def get_file(fname,

return fpath

def _hash_file(fpath, algorithm='sha256', chunk_size=65535):

return hasher.hexdigest()

def validate_file(fpath, file_hash, algorithm='auto', chunk_size=65535):

return False

@keras_export('keras.utils.Sequence')

class Sequence(object):

yield item

def iter_sequence_infinite(seq):

yield item

# Global variables to be shared across processes

_SHARED_SEQUENCES = {}

# We use a Value to provide unique id to different processes.

_SEQUENCE_COUNTER = None

# Because multiprocessing pools are inherently unsafe, starting from a clean

# state can be essential to avoiding deadlocks. In order to accomplish this, we

# need to be able to check on the status of Pools that we create.

_DATA_POOLS = weakref.WeakSet()

_WORKER_ID_QUEUE = None # Only created if needed.

_WORKER_IDS = set()

def get_worker_id_queue():

return _WORKER_ID_QUEUE

def init_pool(seqs):

global _SHARED_SEQUENCES

_SHARED_SEQUENCES = seqs

@keras_export('keras.experimental.terminate_keras_multiprocessing_pools')

def terminate_keras_multiprocessing_pools(grace_period=0.1, use_sigkill=False):

return errors

def get_index(uid, i):

return _SHARED_SEQUENCES[uid][i]

@keras_export('keras.utils.SequenceEnqueuer')

class SequenceEnqueuer(object):

raise NotImplementedError

@keras_export('keras.utils.OrderedEnqueuer')

class OrderedEnqueuer(SequenceEnqueuer):

six.reraise(*sys.exc_info())

def init_pool_generator(gens, random_seed=None, id_queue=None):

id_queue.put(worker_proc.ident, block=True, timeout=0.1)

def next_sample(uid):

return six.next(_SHARED_SEQUENCES[uid])

@keras_export('keras.utils.GeneratorEnqueuer')

class GeneratorEnqueuer(SequenceEnqueuer):

six.reraise(*sys.exc_info())