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

How to use get_cpus method in autotest

Best Python code snippet using autotest_python

run_example.py

Source:run_example.py

...19print("Nr. of CPU cores: ",cpus)20print("-----------------------------------")21def mCPU(func, var, n_jobs=20,verbose=10):22    return Parallel(n_jobs=n_jobs, verbose=verbose)(delayed(func)(i) for i in var)23def get_cpus(data):24    if len(data) < cpus:25        return len(data)26    else:27        return cpus28def get_all_func(FUNC,ARRAY,n_jobs):29    def do_func(A):30        return [FUNC(i) for i in A]31    return mCPU(do_func,ARRAY,n_jobs)32def get_all_func_2(FUNC,A1,A2):33    def do_func(n):34        return [FUNC(A1[n][i],A2[n][i]) for i in range(len(A1[n]))]35    return [do_func(j) for j in tqdm(range(len(A1)))]36if __name__ == '__main__':37    from glob import glob38    image_path = "data/images/"39    image_paths= sorted(glob(image_path+"*"))40    print("-------------------------------------")41    print("Input images:")42    for i in image_paths:43        print(i)44    print("-------------------------------------")45    model_path = "data/models/UNET_weight_state.pt"46    npy_path   = "data/npy/"47    out_path   = "data/out/"48    Path(npy_path).mkdir(parents=True, exist_ok=True)49    Path(out_path).mkdir(parents=True, exist_ok=True)50    size_filter= 60051    hole_size  = 452    print("Load models:")53    model      = models.UNET(1,4)54    model.load_state_dict(torch.load(model_path, map_location=torch.device('cpu')))55    print("inference:")56    def process_tracks(path):57        image      =  io.imread(path)[:,:,0] ### for grey-scale images58        cells      =  inference.cell_inference(image, model, size_filter, hole_size)59        return [image, cells]60    out_arr           = np.asarray(mCPU(process_tracks,image_paths, get_cpus(image_paths)))61    images, cell_mask = np.swapaxes(out_arr,0,1)62    print("Tracking cells:")63    #cell_tracks       = tracking.track_clustering(cell_mask[::-1])64    #cell_tracks[-1]   = tracking.relable_cells(cell_tracks[-1])65    #cell_tracks       = tracking.track_clustering(cell_tracks[::-1])66    M, cell_tracks_float = tracking.track_merger_splits(cell_mask)67    uniques              = ap.find_unique(cell_tracks_float )68    cell_tracks          = ap.get_int_masks(cell_tracks_float,uniques)69    print("Cell activity estimation")70    print("Computing cell vertices")71    cell_vertices = ap.get_all_dist_centers(cell_tracks,get_cpus(cell_tracks))72    np.save(npy_path+"AI_ucents"       ,cell_vertices)73    print("Computing individual cell FOV parameters")74    radius        = ap.get_all_radii(cell_tracks, cell_vertices).max()75    print("Computing persistence")76    RTS           =  np.asarray([ap.get_persistence(cell_vertices,i) for i in tqdm(range(cell_vertices.shape[1]))]) #### needs error estimation77    print("Computing polar transformations")78    cells      = np.asarray([ap.extract_rad_frames(cell_tracks, cell_vertices,int(radius),i) for i in tqdm(uniques)],dtype="bool")79    polts      = np.asarray(get_all_func(ap.pol_trans,cells,get_cpus(cells))).astype(int)80    polins     = np.asarray(get_all_func(ap.pol_outline,polts,get_cpus(polts)))81    ris,ais    = np.asarray(get_all_func(ap.inner_circ,polins,get_cpus(polins))).T82    ros,aos    = np.asarray(get_all_func(ap.outer_max,polins,get_cpus(cells))).T83    LEN_RATIOS = ris/ros84    poltokis   = np.asarray(get_all_func(ap.get_toki,polins,get_cpus(cells)))85    cartokis   = np.asarray(get_all_func_2(ap.get_cart_toki,poltokis.T,cell_vertices)).T86    np.save(npy_path+"images"          ,images)87    np.save(npy_path+"cell_masks_float",cell_tracks_float)88    np.save(npy_path+"cell_masks"      ,cell_tracks)89    np.save(npy_path+"cell_crops"      ,cells)90    np.save(npy_path+"AI_tokis"        ,cartokis)91    np.save(npy_path+"AI_cell_lw_rats,",LEN_RATIOS)...

data_extract.py

Source:data_extract.py

...40        Returns:41            str: test string42    """43    return [x.stem.replace("_"," ").title() for x in get_tests(testbench) if x != None]44def get_cpus(testbench:str=None,test:str=None)->list:45    """ Returns list of cpus. 46        Args:47            None48        Returns:49            list: list of cpus50    """51    if testbench and test:52        test = test.lower().replace(" ","_")53        cpus = [list(x.iterdir()) for x in get_tests(testbench) if test in str(x)][0]54        return cpus55    return None56def get_cpus_str(testbench:str=None,test:list|str=None)->list:57    """ Returns cpu string list.58        Args:59            testbench (str): testbench name60            test (str): test name61        Returns:62            str: cpu string63    """64    if testbench and test:65        return [x.stem for x in get_cpus(testbench,test) if x != None]66def get_data_from_tb(testbench:str=None,test:str=None,tag:str=None,cpu:str=None,stats:str=None):67    """ Extracts data from testbench. 68        Args:69            testbench (str): testbench eg. Baseline Testbench70            test (str): test eg. Rising Session71            cpu (str): cpu eg. 16vCPU64GBmemory72            tag (str): tag73        Returns:74            dict: dictionary with extracted data. eg {x=""}75    """76    if cpu:77        dirs_cpu_tb = get_cpus(testbench,test)78        dirs_data = [list(x.iterdir()) for x in dirs_cpu_tb if x.stem == cpu][0]79        filtered_data = {}80            # filtered_data[cpu_tb.stem]={}81        for dir_data in dirs_data:82            with open(dir_data, 'r') as f:83                raw_data = json.load(f)84                if tag in raw_data:85                    if isinstance(raw_data[tag], dict):86                        data_a=raw_data[tag]["6-1"]["stats"][stats]["values"].replace("[","").replace("]","").split(",")87                        filtered_data[dir_data.stem]=[float(a) for a in data_a]88                    else:89                        filtered_data[dir_data.stem]=raw_data[tag]90        # Make all array same length91        # max_len = max([len(x) for x in filtered_data.values()])...

main.py

Source:main.py

...19def main():20    threading.Timer(SENT_INTERVAL, main).start()21    device = [{22        'id': DEVICE_FRIENDLY_NAME,23        'sensors': get_temperatures() + get_internet_speed() + get_memory() + get_disks() + get_cpus() + get_battery() + get_networks() + get_fans()24        }]25    connection = http.client.HTTPSConnection(UBEAC_URL)26    connection.request('GET', GATEWAY_URL, json.dumps(device))27    response = connection.getresponse()28    print(response.read().decode())293031if __name__ == '__main__':
...

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.