Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_popitem():
mapping = [(val, pos) for pos, val in enumerate(string.ascii_lowercase)]
temp = SortedDict(mapping)
assert temp.popitem() == ('z', 25)
def test_reversed_key():
temp = SortedDict(modulo, ((val, val) for val in range(100)))
temp._reset(7)
values = sorted(range(100), key=modulo)
assert all(lhs == rhs for lhs, rhs in zip(reversed(temp), reversed(values)))
def __init__(self, env, args):
super(UpsideDownRL, self).__init__()
self.env = env
self.args = args
self.nb_actions = self.env.action_space.n
self.state_space = self.env.observation_space.shape[0]
# Use sorted dict to store experiences gathered.
# This helps in fetching highest reward trajectories during exploratory stage.
self.experience = SortedDict()
self.B = BehaviorFunc(self.state_space, self.nb_actions, args).cuda()
self.optimizer = optim.Adam(self.B.parameters(), lr=self.args.lr)
self.use_random_actions = True # True for the first training epoch.
self.softmax = nn.Softmax()
# Used to clip rewards so that B does not get unrealistic expected reward inputs.
self.lunar_lander_max_reward = 250
POWCountOfEpoch = SortedDict()
DSBlockSendTime = SortedDict()
DSBlockRecvTime = SortedDict()
DSLeaderIp = SortedDict()
DSLeaderNodeId = SortedDict()
TxnProcStartTime = SortedDict()
TxnProcEndTime = SortedDict()
MIConsensusLeader = SortedDict()
MIConsensusDict = SortedDict()
MIBlockSendTime = SortedDict()
MIBlockRecvTime = SortedDict()
FBConsensusDict = SortedDict()
FLBlockSendTime = SortedDict()
FLBlockRecvTime = SortedDict()
NodeLastBlock = SortedDict()
NodeReward = SortedDict()
mbConsensusStartTime = None
fbConsensusStartTime = None
LatestBlockNumber = 0
def print_usage():
print ("Copyright (C) Zilliqa\n")
print ("Profile consensus and communication time from state logs\n"
timeSpan = 0
shardId = 0
class Node:
nodeId = 0
ipAddress = ''
DSConsensusStartTime = SortedDict()
DSConsensusEndTime = SortedDict()
POWCountOfEpoch = SortedDict()
DSBlockSendTime = SortedDict()
DSBlockRecvTime = SortedDict()
DSLeaderIp = SortedDict()
DSLeaderNodeId = SortedDict()
TxnProcStartTime = SortedDict()
TxnProcEndTime = SortedDict()
MIConsensusLeader = SortedDict()
MIConsensusDict = SortedDict()
MIBlockSendTime = SortedDict()
MIBlockRecvTime = SortedDict()
FBConsensusDict = SortedDict()
FLBlockSendTime = SortedDict()
FLBlockRecvTime = SortedDict()
NodeLastBlock = SortedDict()
NodeReward = SortedDict()
self.logger.debug("Excluding %s as it has a score <= 0", tid)
self.transcripts[tid].score = 0
self._not_passing.add(tid)
assert self.transcripts[tid].score is not None
if tid in self._not_passing:
pass
else:
assert self.transcripts[tid].score == sum(self.scores[tid].values()), (
tid, self.transcripts[tid].score, sum(self.scores[tid].values())
)
self.scores[tid]["score"] = self.transcripts[tid].score
else:
valid_metrics = self.regressor.metrics
metric_rows = SortedDict()
for tid, transcript in sorted(self.transcripts.items(), key=operator.itemgetter(0)):
for param in valid_metrics:
self.scores[tid][param] = "NA"
row = []
for attr in valid_metrics:
val = getattr(transcript, attr)
if isinstance(val, bool):
if val:
val = 1
else:
val = 0
row.append(val)
# Necessary for sklearn ..
row = numpy.array(row)
# row = row.reshape(1, -1)
metric_rows[tid] = row
def __init__(self):
self.logger = get_logger(self.__class__.__name__)
self.asks = SortedDict()
self.bids = SortedDict()
self.timestamp = 0
def __init__(self, name, timeout, rand=0):
self.db = {}
self.name = name
self.timestamps = sortedcontainers.SortedDict()
self.timeout = timeout
self.rand = rand
self.base_redis_key = "RSSIT:" + str(self.name) + ":"
def rebin(binned, key_function):
result = sortedcontainers.SortedDict()
for bin_start, value in binned.items():
new_bin_start = key_function(bin_start)
try:
result[new_bin_start] += value
except KeyError:
result[new_bin_start] = value
return result
def __init__(self, my_book):
super().__init__(None)
# Save reference to the book
self.my_book = my_book
# Save reference to the metamanager
self.metamgr = my_book.get_meta_manager()
# Save reference to the edited document
self.document = my_book.get_edit_model()
# Save reference to a speller, which will be the default
# at this point.
self.speller = my_book.get_speller()
# The vocabulary list as a sorted dict.
self.vocab = SortedDict()
# Key and Values views on the vocab list for indexing by table row.
self.vocab_kview = self.vocab.keys()
self.vocab_vview = self.vocab.values()
# The count of available words based on the latest sort
self.active_word_count = 0
# The good- and bad-words sets and the scannos set.
self.good_words = set()
self.bad_words = set()
self.scannos = set()
# A dict of words that use an alt-dict tag. The key is a word and the
# value is the alt-dict tag string.
self.alt_tags = SortedDict()
# Cached sort vectors, see get_sort_vector()
self.sort_up_vectors = [None, None, None]
self.sort_down_vectors = [None, None, None]
self.sort_key_funcs = [None, None, None]