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

How to use start_replay method in localstack

Best Python code snippet using localstack_python

train.py

Source:train.py

1import numpy as np2import argparse3from cave import DQNCave4from dqn_agent import DQNAgent5from drqn_agent import DRQNAgent6import copy7from collections import deque8if __name__ == "__main__":9    parser = argparse.ArgumentParser()10    parser.add_argument("-m", "--model_path")11    parser.add_argument("-l", "--load", dest="load", action="store_true", default=False)12    parser.add_argument("-e", "--epoch-num", dest="n_epochs", default=200, type=int)13    parser.add_argument("-s", "--save-interval", dest="save_interval", default=1000, type=int)14    parser.add_argument("-g", "--graves", dest="graves", action="store_true", default=False, help='Use RmpropGraves (default: off)')15    parser.add_argument("-r", "--recurent", dest="drqn", action="store_true", default=False, help='Use DRQN (default: off)')16    parser.add_argument("-d", "--ddqn", dest="ddqn", action="store_true", default=False, help='Use Double DQN (default: off)')17    args = parser.parse_args()18    # parameters19    n_epochs = args.n_epochs20    state_num = 521    # environment, agent22    env = DQNCave()23    Agent = DRQNAgent if args.drqn else DQNAgent24    agent = Agent(env.enable_actions, env.name, env.size, state_num, graves=args.graves, ddqn=args.ddqn)25    agent.init_model()26    if args.load:27        agent.load_model(args.model_path)28    # variables29    e = 030    total_frame = 031    while e < n_epochs:32        # reset33        frame = 034        loss = 0.035        do_replay_count = 036        Q_max = 0.037        env.reset()38        state_t_1, reward_t, terminal, past_time = env.observe()39        S = deque(maxlen=state_num * 2)40        while not terminal:41            state_t = state_t_142            if len(S) == 0:43                [S.append([state_t] if args.drqn else state_t) for i in range(state_num * 2)]44            else:45                S.append([state_t] if args.drqn else state_t)46                # execute action in environment47            tmpS = [S[(s + 1) * 2 - 1] for s in range(state_num)]48            if frame % 3 == 0:49                action_t = agent.select_action(tmpS, agent.exploration)50                #action_t = agent.select_action([state_t], agent.exploration)51            env.execute_action(action_t)52            # observe environment53            state_t_1, reward_t, terminal, past_time = env.observe()54            # store experience55            start_replay = False56            if frame % 4 == 0 or reward_t == -1:57                new_S = copy.copy(S)58                new_S.append([state_t_1] if args.drqn else state_t_1)59                tmpnew_S = [new_S[(s + 1) * 2 - 1] for s in range(state_num)]60                score = past_time if e > 200 else None61                start_replay = agent.store_experience(tmpS, action_t, reward_t, tmpnew_S, terminal, score)62                #start_replay = agent.store_experience([state_t], action_t, reward_t, [state_t_1], terminal)63            # experience replay64            if start_replay:65                do_replay_count += 166                if do_replay_count % 3 == 0 or reward_t == -1:67                    agent.update_exploration(e)68                    agent.experience_replay()69                    do_replay_count = 070            # update target network71            if total_frame % 5000 == 0 and start_replay:72                agent.update_target_model()73            # for log74            frame += 175            total_frame += 176            loss += agent.current_loss77            Q_max += np.max(agent.Q_values(tmpS))78        if start_replay:79            print("EPOCH: {:03d}/{:03d} | SCORE: {:03d} | LOSS: {:.4f} | Q_MAX: {:.4f}".format(80                e, n_epochs - 1, past_time, loss / frame, Q_max / frame))81        if e > 0 and e % args.save_interval == 0:82            agent.save_model(e)83            agent.save_model()84        if start_replay:85            e += 186    # save model...

patch_socket.py

Source:patch_socket.py

...9    def start_record(self):10        self.recording = True11        self.replaying = False12        self.records = []13    def start_replay(self):14        self.recording = False15        self.replaying = True16        self.replay_records = self.records[:]17    def settimeout(self, n):18        return self._ss.settimeout(n)19    def connect(self, *args, **kwargs):20        return self._ss.connect(*args, **kwargs)21    def setsockopt(self, *args):22        if not self.replaying:23            return self._ss.setsockopt(*args)24    def shutdown(self, *args):25        if not self.replaying:26            return self._ss.shutdown(*args)27    def send(self, buf):28        if self.replaying:29            return len(buf)30        else:31            return self._ss.send(buf)32    def sendall(self, buf):33        if self.replaying:34            return len(buf)35        else:36            return self._ss.sendall(buf)37    def recv(self, size):38        if self.replaying:39            return self.replay_records.pop(0)40        buf = self._ss.recv(size)41        if self.recording:42            self.records.append(buf)43        return buf44    def recv_into(self, buf):45        if self.replaying:46            s = self.replay_records.pop(0)47            buf[:len(s)] = s48            return len(s)49        ret = self._ss.recv_into(buf)50        if self.recording:51            self.records.append(buf[:ret])52        return ret53    def close(self):54        return self._ss.close()55import socket as socketmodule56socketmodule.socket = socket57def start_record(socks):58    for sock in socks:59        sock.start_record()60def start_replay(socks):61    for sock in socks:62        sock.start_replay()63def run_with_recording(socks, func):64    for sock in socks:65        sock.start_record()66    func()67def run_with_replay(socks, func):68    for sock in socks:69        sock.start_replay()...

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.