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

How to use num_processes method in autotest

Best Python code snippet using autotest_python

batchsim3.py

Source:batchsim3.py

1import os2import inspect3currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))4parentdir = os.path.dirname(os.path.dirname(currentdir))5os.sys.path.insert(0, parentdir)6from pybullet_utils import bullet_client7from pybullet_envs.examples import panda_sim8import time9useGUI = False10timeStep = 1./60.11# Importing the libraries12import os13import time14import multiprocessing as mp15from multiprocessing import Process, Pipe16pandaEndEffectorIndex = 11 #817pandaNumDofs = 718_RESET = 119_CLOSE = 220_EXPLORE = 321def ExploreWorker(rank, num_processes, childPipe, args):22  print("hi:",rank, " out of ", num_processes)  23  import pybullet as op124  import pybullet_data as pd25  logName=""26  p1=027  n = 028  space = 2 29  simulations=[]30  sims_per_worker = 1031  32  offsetY = rank*space33  while True:34    n += 135    try:36      # Only block for short times to have keyboard exceptions be raised.37      if not childPipe.poll(0.0001):38        continue39      message, payload = childPipe.recv()40    except (EOFError, KeyboardInterrupt):41      break42    if message == _RESET:43      if (useGUI):44        p1 = bullet_client.BulletClient(op1.GUI)45      else:46        p1 = bullet_client.BulletClient(op1.DIRECT)47      p1.setTimeStep(timeStep)48      49      p1.setPhysicsEngineParameter(numSolverIterations=8)50      p1.setPhysicsEngineParameter(minimumSolverIslandSize=100)51      p1.configureDebugVisualizer(p1.COV_ENABLE_Y_AXIS_UP,1)52      p1.configureDebugVisualizer(p1.COV_ENABLE_RENDERING,0)53      p1.setAdditionalSearchPath(pd.getDataPath())54      p1.setGravity(0,-9.8,0)55      logName = str("batchsim")+str(rank)56      for j in range (3):57        offsetX = 0#-sims_per_worker/2.0*space58        for i in range (sims_per_worker):59          offset=[offsetX,0, offsetY]60          sim = panda_sim.PandaSim(p1, offset)61          simulations.append(sim)62          offsetX += space 63        offsetY += space 64      childPipe.send(["reset ok"])65      p1.configureDebugVisualizer(p1.COV_ENABLE_RENDERING,1)66      for i in range (100):67        p1.stepSimulation()68      69      logId = p1.startStateLogging(op1.STATE_LOGGING_PROFILE_TIMINGS,logName)70      continue71    if message == _EXPLORE:72      sum_rewards=rank73      74      if useGUI:75        numSteps = int(20000)76      else:77        numSteps = int(5)78      for i in range (numSteps):79        for s in simulations:80          s.step()81        p1.stepSimulation()82      #print("logId=",logId)83      #print("numSteps=",numSteps)84      childPipe.send([sum_rewards])85      continue86    if message == _CLOSE:87      p1.stopStateLogging(logId)88      childPipe.send(["close ok"])89      break90  childPipe.close()91  92if __name__ == "__main__":93  mp.freeze_support()94  if useGUI:95    num_processes = 196  else:97    num_processes = 1298  processes = []99  args=[0]*num_processes100  101  childPipes = []102  parentPipes = []103  for pr in range(num_processes):104    parentPipe, childPipe = Pipe()105    parentPipes.append(parentPipe)106    childPipes.append(childPipe)107  for rank in range(num_processes):108    p = mp.Process(target=ExploreWorker, args=(rank, num_processes, childPipes[rank],  args))109    p.start()110    processes.append(p)111      112  113  for parentPipe in parentPipes:114    parentPipe.send([_RESET, "blaat"])115  116  positive_rewards = [0]*num_processes117  for k in range(num_processes):118    #print("reset msg=",parentPipes[k].recv()[0])119    msg = parentPipes[k].recv()[0]120  121  for parentPipe in parentPipes:122    parentPipe.send([_EXPLORE, "blaat"])123  124  positive_rewards = [0]*num_processes125  for k in range(num_processes):126    positive_rewards[k] = parentPipes[k].recv()[0]127    #print("positive_rewards=",positive_rewards[k])128  129  for parentPipe in parentPipes:130    parentPipe.send([_EXPLORE, "blaat"])131  132  positive_rewards = [0]*num_processes133  for k in range(num_processes):134    positive_rewards[k] = parentPipes[k].recv()[0]135    #print("positive_rewards=",positive_rewards[k])136    msg = positive_rewards[k]137  for parentPipe in parentPipes:138    parentPipe.send([_EXPLORE, "blaat"])139  140  positive_rewards = [0]*num_processes141  for k in range(num_processes):142    positive_rewards[k] = parentPipes[k].recv()[0]143    #print("positive_rewards=",positive_rewards[k])144  145  for parentPipe in parentPipes:146    parentPipe.send([_CLOSE, "pay2"])147  148  for p in processes:149    p.join()150  151  #now we merge the separate json files into a single one152  fnameout = 'batchsim.json'153  count = 0154  outfile = open(fnameout, "w+")155  outfile.writelines(["{\"traceEvents\":[\n"])156  numFiles = num_processes157  for num in range(numFiles):158    print("num=",num)159    fname = 'batchsim%d_0.json' % (num)160    with open(fname) as infile:161        for line in infile:162          if "pid" in line:163            line = line.replace('\"pid\":1', '\"pid\":'+str(num))164            if num < (numFiles-1) and not "{}}," in line:165              line = line.replace('{}}', '{}},')166              print("line[",count,"]=",line)167            outfile.write(line)168          count += 1169  print ("count=",count)170  outfile.writelines(["],\n"])171  outfile.writelines(["\"displayTimeUnit\": \"ns\"}\n"])...

test_globals.py

Source:test_globals.py

1import os2from copy import deepcopy3from gfw_pixetl.settings.globals import Globals4def test_global_workers():5    config = Globals()6    assert config.num_processes == os.cpu_count()7    assert config.workers == os.cpu_count()8    config.workers = os.cpu_count() + 19    assert config.num_processes == os.cpu_count()10    assert config.workers == os.cpu_count()11    config.num_processes = 112    assert config.num_processes == 113    assert config.workers == 114    config = Globals(num_processes=2, workers=3)15    assert config.num_processes == 216    assert config.workers == 217    config = Globals(cores=4, num_processes=3, workers=2)18    assert config.num_processes == 319    assert config.workers == 220    vars = deepcopy(os.environ)21    os.environ["NUM_PROCESSES"] = "2"22    config = Globals(workers=3)23    if os.cpu_count() >= 2:24        assert config.num_processes == 225        assert config.workers == 226    else:27        assert config.num_processes == 128        assert config.workers == 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.