How to use write_cores method in autotest

Best Python code snippet using autotest_python

crash_handler.py

Source:crash_handler.py

...132                   time.ctime(float(crash_time)))133    report += "Program backtrace:\n%s\n" % backtrace134    report_path = os.path.join(basedir, 'report')135    write_to_file(report_path, report)136def write_cores(core_data, dir_list):137    """138    Write core files to all directories, optionally providing reports.139    @param core_data: Contents of the core file.140    @param dir_list: List of directories the cores have to be written.141    @param report: Whether reports are to be generated for those core files.142    """143    syslog.syslog("Writing core files to %s" % dir_list)144    for result_dir in dir_list:145        if not os.path.isdir(result_dir):146            os.makedirs(result_dir)147        core_path = os.path.join(result_dir, 'core')148        core_path = write_to_file(core_path, core_file, report=True)149if __name__ == "__main__":150    syslog.openlog('AutotestCrashHandler', 0, syslog.LOG_DAEMON)151    global crashed_pid, crash_time, uid, signal, hostname, exe152    try:153        full_functionality = False154        try:155            crashed_pid, crash_time, uid, signal, hostname, exe = sys.argv[1:]156            full_functionality = True157        except ValueError, e:158            # Probably due a kernel bug, we can't exactly map the parameters159            # passed to this script. So we have to reduce the functionality160            # of the script (just write the core at a fixed place).161            syslog.syslog("Unable to unpack parameters passed to the "162                          "script. Operating with limited functionality.")163            crashed_pid, crash_time, uid, signal, hostname, exe = (None, None,164                                                                   None, None,165                                                                   None, None)166        if full_functionality:167            core_dir_name = 'crash.%s.%s' % (exe, crashed_pid)168        else:169            core_dir_name = 'core.%s' % generate_random_string(4)170        # Get the filtered results dir list171        results_dir_list = get_results_dir_list(crashed_pid, core_dir_name)172        # Write the core file to the appropriate directory173        # (we are piping it to this script)174        core_file = sys.stdin.read()175        if (exe is not None) and (crashed_pid is not None):176            syslog.syslog("Application %s, PID %s crashed" % (exe, crashed_pid))177        write_cores(core_file, results_dir_list)178    except Exception, e:...

find_cores.py

Source:find_cores.py

...37    # force.38    r = distance.cdist(z_core, z, "euclidean")39    return rho_sph(r, 6)40    41def write_cores(sim_dir, idxs):42    file_name = path.join(sim_dir, "halos", "cores.dat")43    with open(file_name, "wb") as fp:44        fp.write(struct.pack("qqq", idxs.shape[0],45                             idxs.shape[1], idxs.shape[2]))46        idxs = idxs.flatten()47        idxs = np.asarray(idxs, dtype=np.int32)48        idxs.tofile(fp)49def write_core_positions(sim_dir, x, v):50    file_name = path.join(sim_dir, "halos", "core_pos.dat")51    with open(file_name, "wb") as fp:52        fp.write(struct.pack("qq", x.shape[0], x.shape[1]))53        x, v = x.flatten(), v.flatten()54        x = np.asarray(x, dtype=np.float32)55        v = np.asarray(v, dtype=np.float32)56        x.tofile(fp)57        v.tofile(fp)58        59def core_position(idxs, x):60    return x[idxs[0]]61def core_velocity(idxs, v, scale):62    return np.sqrt(scale)*v[idxs[0]]63def track_cores_snap(info, sim_dir, snap, h, hist, infall_core_idxs,64                     core_idxs, core_x, core_v):65    print("snap", snap)66    base_dir, suite, halo_name = find_infall_cores.parse_sim_dir(sim_dir)    67    param = symlib.parameter_table[suite]68    scale = symlib.scale_factors(sim_dir)69    owner_all = symlib.read_particles(info, sim_dir, snap, "ownership")70    valid_all = symlib.read_particles(info, sim_dir, snap, "valid")71    x_all = symlib.read_particles(info, sim_dir, snap, "x")72    v_all = symlib.read_particles(info, sim_dir, snap, "v")73    for sub_i in range(1, len(h)):74        if snap < hist["merger_snap"][sub_i]:75            continue76        elif snap == hist["merger_snap"][sub_i]:77            n_core = np.min(N_CORE_MAX)78            x, v = x_all[sub_i], v_all[sub_i]79            idx = infall_core_idxs[sub_i,:n_core]80            core_idxs[sub_i,snap,:n_core] = idx81            core_x[sub_i,snap,:] = core_position(idx, x)82            core_v[sub_i,snap,:] = core_velocity(idx, v, scale[snap])83        else:84            owner, valid = owner_all[sub_i], valid_all[sub_i]85            x, v = x_all[sub_i], v_all[sub_i]86            87            n_core = np.sum(core_idxs[sub_i,snap-1,:] >= 0)88            x_core = x[core_idxs[sub_i,snap-1,:n_core]]89            v_core = v[core_idxs[sub_i,snap-1,:n_core]]90            ok = valid & (owner == 0)91            92            idx = np.arange(len(x), dtype=int)[ok]93            rho = phase_space_density(x_core, v_core, x[ok], v[ok])94            95            top_n_idx = np.argpartition(rho, -n_core)[-n_core:]96            top_n_rho = rho[top_n_idx]97            top_n_order = np.argsort(top_n_rho)[::-1]98            idx = idx[top_n_idx[top_n_order]]99            core_idxs[sub_i,snap,:n_core] = idx100            core_x[sub_i,snap,:] = core_position(idx, x)101            core_v[sub_i,snap,:] = core_velocity(idx, v, scale[snap])102            103def track_cores(sim_dir):104    base_dir, suite, halo_name = find_infall_cores.parse_sim_dir(sim_dir)105    106    param = symlib.parameter_table[suite]107    scale = symlib.scale_factors(sim_dir)108    n_snap = len(scale)109        110    h, hist = symlib.read_subhalos(param, sim_dir)111    112    info = symlib.ParticleInfo(sim_dir)113    core_idxs = np.ones((len(h), n_snap, N_CORE_MAX), dtype=int)*-1114    core_x = np.zeros((len(h), n_snap, 3))115    core_v = np.zeros((len(h), n_snap, 3))116    117    infall_core_idxs = symlib.read_particles(info, sim_dir, 0, "infall_core")118    for snap in range(np.min(hist["merger_snap"][1:]), n_snap):119        track_cores_snap(info, sim_dir, snap, h, hist, infall_core_idxs,120                         core_idxs, core_x, core_v)121        122    write_cores(sim_dir, core_idxs)123    write_core_positions(sim_dir, core_x, core_v)124def main():125    config_name, idx_str = sys.argv[1], sys.argv[2]126    target_idx = int(idx_str)127    sim_dirs = find_infall_cores.get_sim_dirs(config_name)128    129    for host_i in range(len(sim_dirs)):130        if host_i != target_idx and target_idx != -1: continue131        sim_dir = sim_dirs[host_i]132        if sim_dir[-1] == "/": sim_dir = sim_dir[:-1]133        track_cores(sim_dir)...

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.