KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use not_done method in Testify

Best Python code snippet using Testify_python

utils.py

Source:utils.py

1import pickle2import numpy as np3import torch4class EpisodicBuffer(object):5    def __init__(self, state_dim, num_actions, buffer_size, horizon, device):6        self.max_size = int(buffer_size)7        self.horizon = horizon8        self.device = device9        self.state = np.zeros((self.max_size, horizon, state_dim))10        self.action = np.zeros((self.max_size, horizon, 1))11        self.reward = np.zeros((self.max_size, horizon, 1))12        self.not_done = np.zeros((self.max_size, horizon, 1))13        self.pibs = np.zeros((self.max_size, horizon, num_actions))14        self.estm_pibs = np.zeros((self.max_size, horizon, num_actions))15        self.nn_action_dist = np.zeros((self.max_size, horizon, num_actions))16    def sample(self, batch_size):17        if batch_size > self.max_size:18            batch_size = self.max_size19        #ind = np.random.randint(0, self.max_size, size=batch_size)20        ind = np.random.choice(self.max_size, batch_size, replace=False)21        return (22            torch.FloatTensor(self.state[ind]).to(self.device),23            torch.LongTensor(self.action[ind]).to(self.device),24            torch.FloatTensor(self.reward[ind]).to(self.device),25            torch.FloatTensor(self.not_done[ind]).to(self.device),26            torch.FloatTensor(self.pibs[ind]).to(self.device),27            torch.FloatTensor(self.estm_pibs[ind]).to(self.device),28            torch.FloatTensor(self.nn_action_dist[ind]).to(self.device)29        )30    def save(self, filename):31        data = {'observations': self.state,32                'actions': self.action,33                'rewards': self.reward,34                'not_done': self.not_done,35                'pibs': self.pibs,36                'estm_pibs': self.estm_pibs,37                'nn_action_dist': self.nn_action_dist,38                }39        with open(filename, 'wb') as f:40            pickle.dump(data, f)41    def load(self, filename, size=-1):42        with open(filename, 'rb') as f:43            data = pickle.load(f)44            self.state = data["observations"]45            self.action = data["actions"]46            self.reward = data["rewards"]47            self.not_done = data["not_done"]48            self.pibs = data["pibs"]49            if "estm_pibs" in data.keys():50                self.estm_pibs = data["estm_pibs"]51            else:52                self.estm_pibs = data["pibs"]53            self.nn_action_dist = data['nn_action_dist']54        self.max_size = self.state.shape[0]55        print(f"Episodic Buffer loaded with {self.max_size} episides.")56# Generic replay buffer for standard gym tasks57class SASRBuffer(object):58    def __init__(self, state_dim, num_actions, batch_size, buffer_size, device):59        self.batch_size = batch_size60        self.max_size = int(buffer_size)61        self.device = device62        self.state = np.zeros((self.max_size, state_dim))63        self.action = np.zeros((self.max_size, 1))64        self.next_state = np.array(self.state)65        self.reward = np.zeros((self.max_size, 1))66        self.not_done = np.zeros((self.max_size, 1))67        self.pibs = np.zeros((self.max_size, num_actions))  # p(a|s) under behavior policy, if we've logged them68        self.nn_action_dist = np.zeros(69            (self.max_size, num_actions))  # dist(s,s') where s' is the nearest neighbor of s with same action70    def sample(self, batch_size=None):71        if batch_size is None:72            batch_size = self.batch_size73        if batch_size > self.max_size:74            batch_size = self.max_size75        ind = np.random.randint(0, self.max_size, size=batch_size)76        return (77            torch.FloatTensor(self.state[ind]).to(self.device),78            torch.LongTensor(self.action[ind]).to(self.device),79            torch.FloatTensor(self.next_state[ind]).to(self.device),80            torch.FloatTensor(self.reward[ind]).to(self.device),81            torch.FloatTensor(self.not_done[ind]).to(self.device),82            torch.FloatTensor(self.pibs[ind]).to(self.device),83            torch.FloatTensor(self.nn_action_dist[ind]).to(self.device)84        )85    def load(self, filename):86        with open(filename, 'rb') as f:87            data = pickle.load(f)88        n_episodes = data["observations"].shape[0]89        horizon = data["observations"].shape[1]90        n = 091        for i in range(n_episodes):92            for h in range(horizon):93                self.state[n, :] = data["observations"][i, h, :]94                self.action[n, 0] = data["actions"][i, h, :]95                self.reward[n, 0] = data["rewards"][i, h, :]96                self.not_done[n, 0] = data["not_done"][i, h, :]97                if self.not_done[n, 0] and h < horizon:98                    self.next_state[n, :] = data["observations"][i, h+1, :]99                    if "estm_pibs" in data.keys():100                        self.pibs[n, :] = data["estm_pibs"][i, h + 1, :]101                    else:102                        self.pibs[n, :] = data["pibs"][i, h + 1, :]103                    self.nn_action_dist[n, :] = data['nn_action_dist'][i, h + 1, :]104                    n += 1105                else:106                    self.next_state[n, :] = data["observations"][i, h, :]107                    self.pibs[n, :] = data["pibs"][i, h, :]108                    self.nn_action_dist[n, :] = data['nn_action_dist'][i, h, :]109                    n += 1110                    break111        self.max_size = n...

Weights.py

Source:Weights.py

1def find_flows(edges):2    flows = []3    for edge in edges:4        top = []5        bot = []6        not_done = []7        done = []8        finished = False9        top.append(edge[0])10        bot.append(edge[1])11        count = 012        while finished == False:13            if count == 0:14                list1 = edges15                count = count + 116            else:17                list1 = not_done18            for edge2 in list1:19                if edge2 not in done:20                    if edge2 != edge:21                            if edge2[0] in top:22                                if edge2 not in done:23                                    top.append(edge2[1])24                                    done.append(edge2)25                                if edge2 in not_done:26                                    not_done.remove(edge2)27                            if edge2[1] in top:28                                if edge2 not in done:29                                    top.append(edge2[0])30                                    done.append(edge2)31                                if edge2 in not_done:32                                    not_done.remove(edge2)33                            if edge2[0] in bot:34                                if edge2 not in done:35                                    bot.append(edge2[1])36                                    done.append(edge2)37                                if edge2 in not_done:38                                    not_done.remove(edge2)39                            if edge2[1] in bot:40                                if edge2 not in done:41                                    bot.append(edge2[0])42                                    done.append(edge2)43                                if edge2 in not_done:44                                    not_done.remove(edge2)45                            else:46                                if edge2 not in not_done:47                                    not_done.append(edge2)48                    if len(done) + 1 == len(edges):49                        finished = True50        print(edge)51        print(top)52        print(bot)53        weight = 054        for node in top:55            if node != edge:56                weight = weight + node[2]57        for node in bot:58            if node != edge:59               weight = weight - node[2]60        flows.append(weight/2.0)61    for n in range(len(edges)):62        edges[n].append(flows[n])...

threaded_mapreduce.py

Source:threaded_mapreduce.py

1from collections import defaultdict2from concurrent.futures import ThreadPoolExecutor as Executor3from time import sleep4def report_progress(futures, tag, callback):5    not_done = 16    done = 07    while not_done > 0:8        not_done = 09        done = 010        for fut in futures:11            if fut.done():12                done +=113            else:14                not_done += 115        sleep(0.5)16        if not_done > 0 and callback:17            callback(tag, done, not_done)18    19def async_map(executor, mapper, data):20    futures = []21    for datum in data:22        futures.append(executor.submit(mapper, datum))23    return futures24def map_less_naive(executor, my_input, mapper):25    map_results = async_map(executor, mapper, my_input)26    return map_results27def map_reduce_less_naive(my_input, mapper, reducer, callback=None):28    with Executor(max_workers=2) as executor:29        futures = async_map(executor, mapper, my_input)30        report_progress(futures, 'map', callback)31        #wait(futures).done32        map_results = map(lambda f: f.result(), futures)33        distributor = defaultdict(list)34        for key, value in map_results:35            distributor[key].append(value)36        futures = async_map(executor, reducer, distributor.items())37        report_progress(futures, 'reduce', callback)38        #wait(futures).done39        results = map(lambda f: f.result(), futures)40    return results41words = filter(lambda x: x!= '', map(lambda x: x.strip().rstrip(), ' '.join(open('text.txt', 'rt', encoding='utf-8').readlines()).split(' ')))42def emitter(word):43    #sleep(10)44    return word, 145counter = lambda emitted: (emitted[0], sum(emitted[1]))46def reporter(tag, done, not_done):47    print(f'Operation {tag}: {done}/{done+not_done}')48words = 'Python is great Python rocks'.split(' ')49a = map_reduce_less_naive(words, emitter, counter, reporter)50for i in sorted(a, key=lambda x: x[1]):51    print(i)52#words = 'Python is great Python rocks'.split(' ')53#with Executor(max_workers=4) as executor:54#    maps = map_less_naive(executor, words, emitter)55#    print(maps[-1])56#    not_done = 157#    while not_done > 0:58#        not_done = 059#        for fut in maps:60#            not_done += 1 if not fut.done() else 061#        sleep(1)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.