How to use dpdata - 10 common examples

def detect_batch_size(batch_size, system=None):
    if type(batch_size) == int:
        return batch_size
    elif batch_size == "auto":
        # automaticcaly set batch size, batch_size = 32 // atom_numb (>=1, <=fram_numb)
        s = dpdata.LabeledSystem(system, fmt='deepmd/npy')
        return int(min( np.ceil(32.0 / float(s["coords"].shape[1]) ), s["coords"].shape[0]))
    else:
        raise RuntimeError("Unsupported batch size")

md_task = []
        f_idx = []
        with open(ii) as fp:
            for ii in fp :
                md_task.append(ii.split()[0])
                f_idx.append(ii.split()[1])
        md_task = md_task[:fp_task_max]
        f_idx = f_idx[:fp_task_max]
        cc = 0
        for tt,ff in zip(md_task, f_idx) :
            traj_file = os.path.join(tt, 'traj', '%d.lammpstrj' % int(ff))
            poscar_file = os.path.join(fp_path, 
                                       'task.%03d.%06d' % (int(sidx), cc), 
                                       'POSCAR')
            cc += 1
            sys0 = dpdata.System(traj_file, fmt = 'lammps/dump', type_map = type_map)
            sys1 = dpdata.System(poscar_file, fmt = 'vasp/poscar')
            test_atom_names(testCase, sys0, sys1)

with open(log_lammps, 'r') as fp:
                if 'Total wall time:' not in fp.read():
                    warnings.warn("lammps not finished " + log_lammps + " skip")
                    return None
                else:
                    fp.seek(0)
                    lines = fp.read().split('\n')
                    for ii in lines:
                        if ("Total number of atoms" in ii) and (not 'print' in ii):
                            natoms = float(ii.split('=')[1].split()[0])
                        if ("Final energy per atoms" in ii) and (not 'print' in ii):
                            epa = float(ii.split('=')[1].split()[0])

                    dump = os.path.join(output_dir, 'dump.relax')
                    contcar = os.path.join(output_dir, 'CONTCAR')
                    d_dump = dpdata.System(dump, fmt='lammps/dump')
                    d_dump.to('vasp/poscar', contcar, frame_idx=-1)
                    force = d_dump['forces']

                    result_dict = {"energy": natoms * epa, "force": list(force[-1].reshape(natoms * 3))}
                    return result_dict

with open(log_lammps, 'r') as fp:
                if 'Total wall time:' not in fp.read():
                    warnings.warn("lammps not finished " + log_lammps + " skip")
                    return None
                else:
                    fp.seek(0)
                    lines = fp.read().split('\n')
                    for ii in lines:
                        if ("Total number of atoms" in ii) and (not 'print' in ii):
                            natoms = float(ii.split('=')[1].split()[0])
                        if ("Final energy per atoms" in ii) and (not 'print' in ii):
                            epa = float(ii.split('=')[1].split()[0])

                    dump = os.path.join(output_dir, 'dump.relax')
                    contcar = os.path.join(output_dir, 'CONTCAR')
                    d_dump = dpdata.System(dump, fmt='lammps/dump')
                    d_dump.to('vasp/poscar', contcar, frame_idx=-1)
                    force = d_dump['forces']

                    result_dict = {"energy": natoms * epa, "force": list(force[-1].reshape(natoms * 3))}
                    return result_dict

f_idx = []
        with open(ii) as fp:
            for ii in fp :
                md_task.append(ii.split()[0])
                f_idx.append(ii.split()[1])
        md_task = md_task[:fp_task_max]
        f_idx = f_idx[:fp_task_max]
        cc = 0
        for tt,ff in zip(md_task, f_idx) :
            traj_file = os.path.join(tt, 'traj', '%d.lammpstrj' % int(ff))
            poscar_file = os.path.join(fp_path, 
                                       'task.%03d.%06d' % (int(sidx), cc), 
                                       'POSCAR')
            cc += 1
            sys0 = dpdata.System(traj_file, fmt = 'lammps/dump', type_map = type_map)
            sys1 = dpdata.System(poscar_file, fmt = 'vasp/poscar')
            test_atom_names(testCase, sys0, sys1)

md_task = []
        f_idx = []
        with open(ii) as fp:
            for ii in fp :
                md_task.append(ii.split()[0])
                f_idx.append(ii.split()[1])
        md_task = md_task[:fp_task_max]
        f_idx = f_idx[:fp_task_max]
        cc = 0
        for tt,ff in zip(md_task, f_idx) :
            traj_file = os.path.join(tt, 'traj', '%d.lammpstrj' % int(ff))
            poscar_file = os.path.join(fp_path,
                                       'task.%03d.%06d' % (int(sidx), cc),
                                       'POSCAR')
            cc += 1
            sys0 = dpdata.System(traj_file, fmt = 'lammps/dump', type_map = type_map)
            sys1 = dpdata.System(poscar_file, fmt = 'vasp/poscar')
            test_atom_names(testCase, sys0, sys1)

def _make_fake_md(idx, md_descript, atom_types, type_map, ele_temp = None) :
    """
    md_descript: list of dimension
                 [n_sys][n_MD][n_frame]
    ele_temp: list of dimension
                 [n_sys][n_MD]
    """
    natoms = len(atom_types)
    ntypes = len(type_map)
    atom_types = np.array(atom_types, dtype = int)
    atom_numbs = [np.sum(atom_types == ii) for ii in range(ntypes)]
    sys = dpdata.System()
    sys.data['atom_names'] = type_map
    sys.data['atom_numbs'] = atom_numbs
    sys.data['atom_types'] = atom_types
    for sidx,ss in enumerate(md_descript) :
        for midx,mm in enumerate(ss) :
            nframes = len(mm)
            cells  = np.random.random([nframes,3,3])
            coords = np.random.random([nframes,natoms,3])
            sys.data['coords'] = coords
            sys.data['cells'] = cells
            task_dir = os.path.join('iter.%06d' % idx,
                                    '01.model_devi',
                                    'task.%03d.%06d' % (sidx, midx))
            os.makedirs(os.path.join(task_dir, 'traj'), exist_ok = True)
            for ii in range(nframes) :
                _write_lammps_dump(sys,

def setUp(self):
        self.places = 5
        self.e_places = 5
        self.f_places = 5
        self.v_places = 2
        assert os.path.isdir('out_data_post_fp_pwscf'), 'out data for post fp pwscf should exist'
        if os.path.isdir('iter.000000') :
            shutil.rmtree('iter.000000')
        shutil.copytree('out_data_post_fp_pwscf', 'iter.000000')
        with open (param_pwscf_file, 'r') as fp :
            jdata = json.load (fp)
        post_fp(0, jdata)
        self.system_1 = dpdata.LabeledSystem('iter.000000/orig', fmt = 'deepmd/raw')
        self.system_2 = dpdata.LabeledSystem('iter.000000/02.fp/data.000', fmt = 'deepmd/raw')

def test_coll(self):

        with open (param_file, 'r') as fp :
            jdata = json.load (fp)
        jdata['out_dir'] = self.odir
        coll_vasp_md(jdata)
        
        sys = dpdata.LabeledSystem(self.odir + '/02.md/sys-004/deepmd//', fmt = 'deepmd/raw')
        self.assertEqual(sys.get_nframes(), 2)
        
        if sys.data['coords'][0][1][0] < sys.data['coords'][1][1][0]:
            idx = [1, 0]
        else :
            idx = [0, 1]
        ref_coord = self.ref_coord[idx]
        ref_cell = self.ref_cell[idx]
        ref_e = self.ref_e[idx]
        ref_f = self.ref_f[idx]
        ref_v = self.ref_v[idx]
        ref_at = self.ref_at
            
        for ff in range(2) :
            self.assertAlmostEqual(ref_e[ff], sys.data['energies'][ff])
        for ii in range(2) :

def setUp (self) :
        type_map = ['C', 'H']
        jdata={
            "cluster_cutoff": 3.5
        }
        self.system_1 = take_cluster("cluster/14400.lammpstrj", type_map, 1125, jdata)
        self.system_2 = dpdata.System.load("cluster/cluster1.json")
        self.places=0