def least_common_values(counter, to_find=None):
if to_find is None:
return sorted(counter.items(), key=itemgetter(1), reverse=False)
return heapq.nsmallest(to_find, counter.items(), key=itemgetter(1))
# build word vector, type: word embedding
python类nsmallest()的实例源码
def _gex_stats(bot, args, author_id, thread_id, thread_type):
data = CARD_TO_USER_ID
number_of_top_cards_to_list = len(data) // 5
number_of_bottom_cards_to_list = len(data) // 10
num_owners = {}
num_in_circulation = {}
for card in data:
num_owners[card] = len(data[card])
num_in_circulation[card] = sum(q[1] for q in data[card])
most_owned_cards = heapq.nlargest(
number_of_top_cards_to_list, num_owners, key=lambda x: num_owners[x])
most_circulated_cards = heapq.nlargest(
number_of_top_cards_to_list, num_in_circulation, key=lambda x: num_in_circulation[x])
least_owned_cards = heapq.nsmallest(
number_of_bottom_cards_to_list, num_owners, key=lambda x: num_owners[x])
least_circulated_cards = heapq.nsmallest(
number_of_bottom_cards_to_list, num_in_circulation, key=lambda x: num_in_circulation[x])
highest_scrabble_scores = heapq.nlargest(
number_of_top_cards_to_list, data, key=lambda x: _get_scrabble_score(x))
out = 'Cards owned by the most people (top 20%):'
FORMAT = '\n{} ({})'
for card in most_owned_cards:
out += FORMAT.format(card, num_owners[card])
bot.sendMessage(out, thread_id=thread_id, thread_type=thread_type)
out = 'Cards with the most copies in circulation (top 20%):'
for card in most_circulated_cards:
out += FORMAT.format(card, num_in_circulation[card])
bot.sendMessage(out, thread_id=thread_id, thread_type=thread_type)
out = 'Cards owned by the fewest people (bottom 10%):'
for card in least_owned_cards:
out += FORMAT.format(card, num_owners[card])
bot.sendMessage(out, thread_id=thread_id, thread_type=thread_type)
out = 'Cards with the fewest copies in circulation (bottom 10%):'
for card in least_circulated_cards:
out += FORMAT.format(card, num_in_circulation[card])
bot.sendMessage(out, thread_id=thread_id, thread_type=thread_type)
out = 'Cards with the highest scrabble score for their ID (top 20%):'
for card in highest_scrabble_scores:
out += FORMAT.format(card, _get_scrabble_score(card))
bot.sendMessage(out, thread_id=thread_id, thread_type=thread_type)
def beam_search(self,initial_state):
'''
Beam search is a graph search algorithm! So I use graph search abstraction
Args:
initial state: an initial stete, python tuple (hx,cx,path,cost)
each state has
hx: hidden states
cx: cell states
path: word indicies so far as a python list e.g. initial is self.token2index["<sos>"]
cost: negative log likelihood
Returns:
captions sorted by the cost (i.e. negative log llikelihood)
'''
found_paths=[]
top_k_states=[initial_state]
while (len(found_paths) < self.beamsize):
#forward one step for all top k states, then only select top k after that
new_top_k_states=[]
for state in top_k_states:
#examine to next five possible states
hy, cy, k_best_next_states = self.successor(state)
for next_state in k_best_next_states:
new_top_k_states.append(next_state)
selected_top_k_states=heapq.nsmallest(self.beamsize, new_top_k_states, key=lambda x : x["cost"])
#within the selected states, let's check if it is terminal or not.
top_k_states=[]
for state in selected_top_k_states:
#is goal state? -> yes, then end the search
if state["path"][-1] == self.token2index["<eos>"] or len(state["path"])==self.depth_limit:
found_paths.append(state)
else:
top_k_states.append(state)
return sorted(found_paths, key=lambda x: x["cost"])
def _gen_loopblocking_perprocess(
nested_loop_desc, resource, cost, part_occ, options,
gen_tifm, gen_tofm, gen_tbat, gen_ords):
def _gen_bl_ts():
'''
Generator for blocking factors.
Transpose LoopEnum-major to BL-major.
'''
gen_lp_ts = [None] * le.NUM
gen_lp_ts[le.IFM] = gen_tifm
gen_lp_ts[le.OFM] = gen_tofm
gen_lp_ts[le.BAT] = gen_tbat
for lp_ts in itertools.product(*gen_lp_ts):
bl_ts = tuple(zip(*lp_ts))
yield bl_ts
def _sweep():
''' Sweep all. '''
is_conv_loops = _is_conv_loops(nested_loop_desc)
for bl_ts, bl_ords in itertools.product(_gen_bl_ts(), gen_ords):
if is_conv_loops and skip_conv(bl_ts, bl_ords):
continue
lbs = LoopBlockingScheme(
nested_loop_desc, bl_ts, bl_ords, resource, part_occ, options)
yield lbs
return heapq.nsmallest(options.ntops, _sweep(),
key=lambda lbs: lbs.get_cost(cost))
def lfu_cache(maxsize=100):
'''Least-frequenty-used cache decorator.
Arguments to the cached function must be hashable.
Cache performance statistics stored in f.hits and f.misses.
Clear the cache with f.clear().
http://en.wikipedia.org/wiki/Least_Frequently_Used
'''
def decorating_function(user_function):
cache = {} # mapping of args to results
use_count = Counter() # times each key has been accessed
kwd_mark = object() # separate positional and keyword args
@functools.wraps(user_function)
def wrapper(*args, **kwds):
key = args
if kwds:
key += (kwd_mark,) + tuple(sorted(kwds.items()))
use_count[key] += 1
# get cache entry or compute if not found
try:
result = cache[key]
wrapper.hits += 1
except KeyError:
result = user_function(*args, **kwds)
cache[key] = result
wrapper.misses += 1
# purge least frequently used cache entry
if len(cache) > maxsize:
for key, _ in nsmallest(maxsize // 10,
use_count.iteritems(),
key=itemgetter(1)):
del cache[key], use_count[key]
return result
def clear():
cache.clear()
use_count.clear()
wrapper.hits = wrapper.misses = 0
wrapper.hits = wrapper.misses = 0
wrapper.clear = clear
return wrapper
return decorating_function
def lru_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least recently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
lastused = {}
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(lastused),
key=itemgetter(1)):
del data[k]
del lastused[k]
data[args] = user_function(*args)
result = data[args]
finally:
lastused[args] = time()
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
lastused.clear()
stats[0] = stats[1] = 0
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least frequently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
usecount = Counter()
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(usecount),
key=itemgetter(1)):
del data[k]
del usecount[k]
data[args] = user_function(*args)
result = data[args]
finally:
usecount[args] += 1
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
usecount.clear()
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
'''Least-frequenty-used cache decorator.
Arguments to the cached function must be hashable.
Cache performance statistics stored in f.hits and f.misses.
Clear the cache with f.clear().
http://en.wikipedia.org/wiki/Least_Frequently_Used
'''
def decorating_function(user_function):
cache = {} # mapping of args to results
use_count = Counter() # times each key has been accessed
kwd_mark = object() # separate positional and keyword args
@functools.wraps(user_function)
def wrapper(*args, **kwds):
key = args
if kwds:
key += (kwd_mark,) + tuple(sorted(kwds.items()))
use_count[key] += 1
# get cache entry or compute if not found
try:
result = cache[key]
wrapper.hits += 1
except KeyError:
result = user_function(*args, **kwds)
cache[key] = result
wrapper.misses += 1
# purge least frequently used cache entry
if len(cache) > maxsize:
for key, _ in nsmallest(maxsize // 10,
use_count.iteritems(),
key=itemgetter(1)):
del cache[key], use_count[key]
return result
def clear():
cache.clear()
use_count.clear()
wrapper.hits = wrapper.misses = 0
wrapper.hits = wrapper.misses = 0
wrapper.clear = clear
return wrapper
return decorating_function
def lru_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least recently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
lastused = {}
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(lastused),
key=itemgetter(1)):
del data[k]
del lastused[k]
data[args] = user_function(*args)
result = data[args]
finally:
lastused[args] = time()
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
lastused.clear()
stats[0] = stats[1] = 0
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least frequently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
usecount = Counter()
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(usecount),
key=itemgetter(1)):
del data[k]
del usecount[k]
data[args] = user_function(*args)
result = data[args]
finally:
usecount[args] += 1
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
usecount.clear()
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def evolve(pop, gamesFactor=2, retain=0.2, random_select=0.05, mutate=0.01):
# Determine the parents to breed from the population
agent_score = {}
numGames = len(pop) * gamesFactor
bar = progressbar.ProgressBar()
for game in bar(range(numGames)):
competitors = random.sample(pop, 2)
game = Board(competitors[0], competitors[1])
winner, history, outcome = game.play()
competitors.remove(winner)
loser = competitors[0]
if winner not in agent_score.keys():
agent_score[winner] = 1
else:
agent_score[winner] += 1
if loser not in agent_score.keys():
agent_score[winner] = -1
else:
agent_score[loser] -= 1
top_performers_size = int(retain * len(pop))
bottom_performers_size = len(pop) - top_performers_size
rand_select_size = int(len(pop) * random_select)
top_perfomers = heapq.nlargest(top_performers_size, agent_score, key=agent_score.get)
bottom_performers = heapq.nsmallest(bottom_performers_size, agent_score, key=agent_score.get)
parents = top_perfomers + random.sample(bottom_performers, rand_select_size)
random.shuffle(parents)
# Create children
numChildren = len(pop) - len(parents)
children = []
for i in range(numChildren):
par = random.sample(parents, 2)
father = par[0]
mother = par[1]
child = breed(mother, father)
children.append(child)
new_pop = parents + children
mutated_pop = []
# Randomly mutate some of the new population
for agent in new_pop:
if mutate > random.uniform(0,1):
print('Mutate')
mutated_agent = mutate_agent(agent)
mutated_pop.append(mutated_agent)
else:
mutated_pop.append(agent)
return mutated_pop
def lru_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least recently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
lastused = {}
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(lastused),
key=itemgetter(1)):
del data[k]
del lastused[k]
data[args] = user_function(*args)
result = data[args]
finally:
lastused[args] = time()
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
lastused.clear()
stats[0] = stats[1] = 0
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least frequently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
usecount = Counter()
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(usecount),
key=itemgetter(1)):
del data[k]
del usecount[k]
data[args] = user_function(*args)
result = data[args]
finally:
usecount[args] += 1
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
usecount.clear()
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def interpolateFromPdfSamples(pdfSamples, x):
xSamples = sorted([i[0] for i in pdfSamples])
if((x<= max(xSamples))and(x >= min(xSamples))):
if(x in xSamples):
## we have an exact match, so we can just return the exact p
## value that was sampled at this point
for xySample in pdfSamples:
if(xySample[0] == x):
return xySample[1]
else:
## the x value we want to sample the probability of falls in
## the range of x values that we have a p value for, so we
## linearly interpolate between the two closest pairs of
## values in x
below = [xySample for xySample in pdfSamples if (xySample[0] < x)]
above = [xySample for xySample in pdfSamples if (xySample[0] > x)]
ptBelow = [xySample for xySample in below if (xySample[0] == max([pt[0] for pt in below]))]
ptBelow =[item for sublist in ptBelow for item in sublist]
ptAbove = [xySample for xySample in above if (xySample[0] == min([pt[0] for pt in above]))]
ptAbove =[item for sublist in ptAbove for item in sublist]
m = (ptAbove[1]-ptBelow[1])/(ptAbove[0]-ptBelow[0])
## calculate slope in this interval
output = ptBelow[1] + m*(x- ptBelow[0])
return output
else:
## the x point in question is beyond the margins that we sampled
## from the pdf, so we linearly interpolate based on the last
## two endpoints
if(x > max(xSamples)):
secondBiggestX = min(heapq.nlargest(2, xSamples))
largest = [xySample for xySample in pdfSamples if (xySample[0] > secondBiggestX)]
largest = [item for sublist in largest for item in sublist]
secondLargest = [xySample for xySample in pdfSamples if (xySample[0] == secondBiggestX)]
secondLargest = [item for sublist in secondLargest for item in sublist]
m = (largest[1]-secondLargest[1])/(largest[0]-secondLargest[0])
## calculate slope in this interval
output = largest[1] + m*(x- largest[0])
elif(x < min(xSamples)):
secondSmallestX = max(heapq.nsmallest(2, xSamples))
smallest = [xySample for xySample in pdfSamples if (xySample[0] < secondSmallestX)]
smallest = [item for sublist in smallest for item in sublist]
secondSmallest = [xySample for xySample in pdfSamples if (xySample[0] == secondSmallestX)]
secondSmallest = [item for sublist in secondSmallest for item in sublist]
m = (secondSmallest[1]-smallest[1])/(secondSmallest[0]-smallest[0])
## calculate slope in this interval
output = smallest[1] + m*(x- smallest[0])
return output
def lru_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least recently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
lastused = {}
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(lastused),
key=itemgetter(1)):
del data[k]
del lastused[k]
data[args] = user_function(*args)
result = data[args]
finally:
lastused[args] = time()
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
lastused.clear()
stats[0] = stats[1] = 0
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least frequently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
usecount = Counter()
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(usecount),
key=itemgetter(1)):
del data[k]
del usecount[k]
data[args] = user_function(*args)
result = data[args]
finally:
usecount[args] += 1
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
usecount.clear()
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
'''Least-frequenty-used cache decorator.
Arguments to the cached function must be hashable.
Cache performance statistics stored in f.hits and f.misses.
Clear the cache with f.clear().
http://en.wikipedia.org/wiki/Least_Frequently_Used
'''
def decorating_function(user_function):
cache = {} # mapping of args to results
use_count = Counter() # times each key has been accessed
kwd_mark = object() # separate positional and keyword args
@functools.wraps(user_function)
def wrapper(*args, **kwds):
key = args
if kwds:
key += (kwd_mark,) + tuple(sorted(kwds.items()))
use_count[key] += 1
# get cache entry or compute if not found
try:
result = cache[key]
wrapper.hits += 1
except KeyError:
result = user_function(*args, **kwds)
cache[key] = result
wrapper.misses += 1
# purge least frequently used cache entry
if len(cache) > maxsize:
for key, _ in nsmallest(maxsize // 10,
use_count.iteritems(),
key=itemgetter(1)):
del cache[key], use_count[key]
return result
def clear():
cache.clear()
use_count.clear()
wrapper.hits = wrapper.misses = 0
wrapper.hits = wrapper.misses = 0
wrapper.clear = clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
'''Least-frequenty-used cache decorator.
Arguments to the cached function must be hashable.
Cache performance statistics stored in f.hits and f.misses.
Clear the cache with f.clear().
http://en.wikipedia.org/wiki/Least_Frequently_Used
'''
def decorating_function(user_function):
cache = {} # mapping of args to results
use_count = Counter() # times each key has been accessed
kwd_mark = object() # separate positional and keyword args
@functools.wraps(user_function)
def wrapper(*args, **kwds):
key = args
if kwds:
key += (kwd_mark,) + tuple(sorted(kwds.items()))
use_count[key] += 1
# get cache entry or compute if not found
try:
result = cache[key]
wrapper.hits += 1
except KeyError:
result = user_function(*args, **kwds)
cache[key] = result
wrapper.misses += 1
# purge least frequently used cache entry
if len(cache) > maxsize:
for key, _ in nsmallest(maxsize // 10,
use_count.iteritems(),
key=itemgetter(1)):
del cache[key], use_count[key]
return result
def clear():
cache.clear()
use_count.clear()
wrapper.hits = wrapper.misses = 0
wrapper.hits = wrapper.misses = 0
wrapper.clear = clear
return wrapper
return decorating_function
def lru_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least recently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
lastused = {}
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(lastused),
key=itemgetter(1)):
del data[k]
del lastused[k]
data[args] = user_function(*args)
result = data[args]
finally:
lastused[args] = time()
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
lastused.clear()
stats[0] = stats[1] = 0
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
def lfu_cache(maxsize=100):
"""A simple cache that, when the cache is full, deletes the least frequently
used 10% of the cached values.
This function duplicates (more-or-less) the protocol of the
``functools.lru_cache`` decorator in the Python 3.2 standard library.
Arguments to the cached function must be hashable.
View the cache statistics tuple ``(hits, misses, maxsize, currsize)``
with f.cache_info(). Clear the cache and statistics with f.cache_clear().
Access the underlying function with f.__wrapped__.
"""
def decorating_function(user_function):
stats = [0, 0] # Hits, misses
data = {}
usecount = Counter()
@functools.wraps(user_function)
def wrapper(*args):
try:
result = data[args]
stats[0] += 1 # Hit
except KeyError:
stats[1] += 1 # Miss
if len(data) == maxsize:
for k, _ in nsmallest(maxsize // 10 or 1,
iteritems(usecount),
key=itemgetter(1)):
del data[k]
del usecount[k]
data[args] = user_function(*args)
result = data[args]
finally:
usecount[args] += 1
return result
def cache_info():
return stats[0], stats[1], maxsize, len(data)
def cache_clear():
data.clear()
usecount.clear()
wrapper.cache_info = cache_info
wrapper.cache_clear = cache_clear
return wrapper
return decorating_function
