Is it possible to convert deepmd/npy or deepmd/raw format to quip/gap/xyz format ?

[Amitcuhp]

Convert deepmd/npy or deepmd/raw format to quip/gap/xyz format

Help will be really appreciated :)

import dpdata
dp_multi_systems = dpdata.MultiSystems.from_dir(
dir_name="./dp/", file_name="./Al*", fmt="deepmd/raw"
)
print(dp_multi_systems)
import os
from dpdata import LabeledSystem
output_dir = "./xyz"
os.makedirs(output_dir, exist_ok=True)
for idx, system in enumerate(dp_multi_systems.systems.values()):
output_path = os.path.join(output_dir, f"system_{idx}.xyz")
try:
system.to(fmt="xyz", file_name=output_path)
print(f"Saved {output_path}")
except Exception as e:
print(f"Failed to save {output_path}: {e}")

it is able to convert only the coordinates. can it convert all?

[Amitcuhp]

Able to convert raw format to quip/gap/xyz

Thank you

[BianTieyuan]

Hi, how to achieve this funciton?

[Amitcuhp]

import os
import re
from collections import OrderedDict
import numpy as np
class QuipGapxyzSystems:
def init(self, raw_dir: str):
self.raw_dir = raw_dir
self.box = self.read_box(os.path.join(raw_dir, 'box.raw'))
self.coords = self.read_coords(os.path.join(raw_dir, 'coord.raw'))
self.energies = self.read_energies(os.path.join(raw_dir, 'energy.raw'))
self.forces = self.read_forces(os.path.join(raw_dir, 'force.raw'))
self.atom_types = self.read_atom_types(os.path.join(raw_dir, 'type.raw'))
self.type_map = self.read_type_map(os.path.join(raw_dir, 'type_map.raw'))
print(f"Length of box: {len(self.box)}")
print(f"Length of coords: {len(self.coords)}")
print(f"Length of energies: {len(self.energies)}")
print(f"Length of forces: {len(self.forces)}")
print(f"Length of atom_types: {len(self.atom_types)}")
print(f"Length of type_map: {len(self.type_map)}")
self.num_atoms = len(self.atom_types)
self.num_frames = len(self.coords) // (self.num_atoms * 3)
print(f"Number of atoms: {self.num_atoms}")
print(f"Number of frames: {self.num_frames}\n")
self.block_generator = self.get_block_generator()
def iter(self):
return self
def next(self):
return self.handle_single_xyz_frame(next(self.block_generator))
def read_box(self, box_file):
with open(box_file, 'r') as f:
lines = f.readlines()
box = []
for line in lines:
box.extend([float(x) for x in line.split()])
return box
def read_coords(self, coord_file):
with open(coord_file, 'r') as f:
lines = f.readlines()
coords = []
for line in lines:
coords.extend([float(x) for x in line.split()])
return coords
def read_energies(self, energy_file):
with open(energy_file, 'r') as f:
energies = [float(line.strip()) for line in f.readlines()]
return energies
def read_forces(self, force_file):
with open(force_file, 'r') as f:
lines = f.readlines()
forces = []
for line in lines:
forces.extend([float(x) for x in line.split()])
return forces
def read_atom_types(self, type_file):
with open(type_file, 'r') as f:
atom_types = [line.strip() for line in f.readlines()]
return atom_types
def read_type_map(self, type_map_file):
with open(type_map_file, 'r') as f:
type_map = [line.strip().split() for line in f.readlines() if line.strip()]
return type_map
def get_block_generator(self):
for frame_idx in range(self.num_frames):
lines = []
lines.append(f"{self.num_atoms}\n")
lines.append(f"Lattice="{self.box[0]:.6f} {self.box[1]:.6f} {self.box[2]:.6f} {self.box[3]:.6f} {self.box[4]:.6f} {self.box[5]:.6f} {self.box[6]:.6f} {self.box[7]:.6f} {self.box[8]:.6f}" Properties=species:S:1:pos:R:3:force:R:3 energy={self.energies[frame_idx]:.6f} pbc="T T T"\n")
for atom_idx in range(self.num_atoms):
coord_idx = frame_idx * self.num_atoms * 3 + atom_idx * 3
force_idx = frame_idx * self.num_atoms * 3 + atom_idx * 3
element_symbol = self.atom_types[atom_idx]
atom_type = self.type_map.index(element_symbol) if element_symbol in self.type_map else 0
if element_symbol == '0':
element_symbol = self.type_map[0][0] # Replace with the first entry from type_map
elif element_symbol == '1':
element_symbol = self.type_map[1][0] # Replace with the second entry from type_map
elif element_symbol == '2':
element_symbol = self.type_map[2][0] # Replace with the third entry from type_map
elif element_symbol == '3':
element_symbol = self.type_map[3][0] # Replace with the fourth entry from type_map
elif element_symbol == '4':
element_symbol = self.type_map[4][0] # Replace with the fourth entry from type_map
elif element_symbol == '5':
element_symbol = self.type_map[5][0] # Replace with the first entry from type_map
elif element_symbol == '6':
element_symbol = self.type_map[6][0] # Replace with the second entry from type_map
elif element_symbol == '7':
element_symbol = self.type_map[7][0] # Replace with the third entry from type_map
elif element_symbol == '8':
element_symbol = self.type_map[8][0] # Replace with the fourth entry from type_map
elif element_symbol == '9':
element_symbol = self.type_map[9][0] # Replace with the fourth entry from type_map
lines.append(f"{element_symbol} {self.coords[coord_idx]:.6f} {self.coords[coord_idx + 1]:.6f} {self.coords[coord_idx + 2]:.6f} "
f"{self.forces[force_idx]:.6f} {self.forces[force_idx + 1]:.6f} {self.forces[force_idx + 2]:.6f}\n")
yield lines
def write_blocks_to_file(self, output_file):
with open(output_file, 'w') as f:
for frame_idx in range(self.num_frames):
lines = []
lines.append(f"{self.num_atoms}\n")
lines.append(f"Lattice="{self.box[0]:.6f} {self.box[1]:.6f} {self.box[2]:.6f} "
f"{self.box[3]:.6f} {self.box[4]:.6f} {self.box[5]:.6f} "
f"{self.box[6]:.6f} {self.box[7]:.6f} {self.box[8]:.6f}" "
f"Properties=species:S:1:pos:R:3:force:R:3 "
f"energy={self.energies[frame_idx]:.6f} pbc="T T T"\n")
for atom_idx in range(self.num_atoms):
coord_idx = frame_idx * self.num_atoms * 3 + atom_idx * 3
force_idx = frame_idx * self.num_atoms * 3 + atom_idx * 3
element_symbol = self.atom_types[atom_idx]
if element_symbol == '0':
element_symbol = self.type_map[0][0] # Replace with the first entry from type_map
elif element_symbol == '1':
element_symbol = self.type_map[1][0] # Replace with the second entry from type_map
elif element_symbol == '2':
element_symbol = self.type_map[2][0] # Replace with the third entry from type_map
elif element_symbol == '3':
element_symbol = self.type_map[3][0] # Replace with the fourth entry from type_map
elif element_symbol == '4':
element_symbol = self.type_map[4][0] # Replace with the fourth entry from type_map
elif element_symbol == '5':
element_symbol = self.type_map[5][0] # Replace with the first entry from type_map
elif element_symbol == '6':
element_symbol = self.type_map[6][0] # Replace with the second entry from type_map
elif element_symbol == '7':
element_symbol = self.type_map[7][0] # Replace with the third entry from type_map
elif element_symbol == '8':
element_symbol = self.type_map[8][0] # Replace with the fourth entry from type_map
elif element_symbol == '9':
element_symbol = self.type_map[9][0] # Replace with the fourth entry from type_map
lines.append(f"{element_symbol} {self.coords[coord_idx]:.6f} "
f"{self.coords[coord_idx + 1]:.6f} {self.coords[coord_idx + 2]:.6f} "
f"{self.forces[force_idx]:.6f} {self.forces[force_idx + 1]:.6f} "
f"{self.forces[force_idx + 2]:.6f}\n")
f.writelines(lines)
@staticmethod
def handle_single_xyz_frame(lines):
atom_num = int(lines[0].strip("\n").strip())
if len(lines) != atom_num + 2:
raise RuntimeError(
f"format error, atom_num=={atom_num}, {len(lines)}!=atom_num+2"
)
data_format_line = lines[1].strip("\n").strip() + " "
field_value_pattern = re.compile(
r"(?P\S+)=(?P['"]?)(?P.*?)(?P=quote)\s+"
)
prop_pattern = re.compile(
r"(?P\w+?):(?P[a-zA-Z]):(?P\d+)"
)
data_format_list = [
kv_dict.groupdict()
for kv_dict in field_value_pattern.finditer(data_format_line)
]
field_dict = {}
for item in data_format_list:
field_dict[item["key"]] = item["value"]

    Properties = field_dict["Properties"]
    prop_list = [
        kv_dict.groupdict() for kv_dict in prop_pattern.finditer(Properties)
    ]
    data_lines = []
    for line in lines[2:]:
        data_lines.append(list(filter(bool, line.strip().split())))
    data_array = np.array(data_lines)
    used_colomn = 0
    type_array = None
    coords_array = None
    Z_array = None
    force_array = None
    virials = None
    for kv_dict in prop_list:
        if kv_dict["key"] == "species":
            if kv_dict["datatype"] != "S":
                raise RuntimeError(
                    "datatype for species must be 'S' instead of {}".format(
                        kv_dict["datatype"]
                    )
                )
            field_length = int(kv_dict["value"])
            type_array = data_array[
                :, used_colomn : used_colomn + field_length
            ].flatten()
            used_colomn += field_length
            continue
        elif kv_dict["key"] == "pos":
            if kv_dict["datatype"] != "R":
                raise RuntimeError(
                    "datatype for pos must be 'R' instead of {}".format(
                        kv_dict["datatype"]
                    )
                )
            field_length = int(kv_dict["value"])
            coords_array = data_array[:, used_colomn : used_colomn + field_length]
            used_colomn += field_length
            continue
        elif kv_dict["key"] == "Z":
            if kv_dict["datatype"] != "I":
                raise RuntimeError(
                    "datatype for pos must be 'R' instead of {}".format(
                        kv_dict["datatype"]
                    )
                )
            field_length = int(kv_dict["value"])
            Z_array = data_array[
                :, used_colomn : used_colomn + field_length
            ].flatten()
            used_colomn += field_length
            continue
        elif kv_dict["key"] == "force":
            if kv_dict["datatype"] != "R":
                raise RuntimeError(
                    "datatype for pos must be 'R' instead of {}".format(
                        kv_dict["datatype"]
                    )
                )
            field_length = int(kv_dict["value"])
            force_array = data_array[:, used_colomn : used_colomn + field_length]
            used_colomn += field_length
            continue
        else:
            raise RuntimeError("unknown field {}".format(kv_dict["key"]))

    type_num_dict = OrderedDict()
    atom_type_list = []
    type_map = {}
    temp_atom_max_index = 0
    if type_array is None:
        raise RuntimeError("type_array can't be None type, check .xyz file")
    for ii in type_array:
        if ii not in type_map:
            type_map[ii] = temp_atom_max_index
            temp_atom_max_index += 1
            temp_atom_index = type_map[ii]
            atom_type_list.append(temp_atom_index)
            type_num_dict[ii] = 1
        else:
            temp_atom_index = type_map[ii]
            atom_type_list.append(temp_atom_index)
            type_num_dict[ii] += 1
    type_num_list = []
    for atom_type, atom_num in type_num_dict.items():
        type_num_list.append((atom_type, atom_num))
    type_num_array = np.array(type_num_list)
    if field_dict.get("virial", None):
        virials = np.array(
            [
                np.array(
                    list(filter(bool, field_dict["virial"].split(" ")))
                ).reshape(3, 3)
            ]
        ).astype("float32")
    else:
        virials = None
    info_dict = {}
    info_dict["atom_names"] = list(type_num_array[:, 0])
    info_dict["atom_numbs"] = list(type_num_array[:, 1].astype(int))
    info_dict["atom_types"] = np.array(atom_type_list).astype(int)
    info_dict["cells"] = np.array(
        [
            np.array(list(filter(bool, field_dict["Lattice"].split(" ")))).reshape(
                3, 3
            )
        ]
    ).astype("float32")
    info_dict["coords"] = np.array([coords_array]).astype("float32")
    info_dict["energies"] = np.array([field_dict["energy"]]).astype("float32")
    info_dict["forces"] = np.array([force_array]).astype("float32")
    if virials is not None:
        info_dict["virials"] = virials
    info_dict["orig"] = np.zeros(3)
    return info_dict

def process_all_directories(base_dir, output_dir):
if not os.path.exists(output_dir):
os.makedirs(output_dir)
for subdir in os.listdir(base_dir):
subdir_path = os.path.join(base_dir, subdir)
if os.path.isdir(subdir_path):
quip_systems = QuipGapxyzSystems(subdir_path)
output_file = os.path.join(output_dir, f"{subdir}.xyz")
quip_systems.write_blocks_to_file(output_file)
input_foldername = input("Enter the input folder name where all raw files located: ")
base_dir = input_foldername
output_foldername = input("Enter the output folder name: ")
output_dir = output_foldername
process_all_directories(base_dir, output_dir)

import os
from ase.io import read, write

def merge_and_sort_xyz_files(folder_path, output_file):
all_atoms = []
for file_name in os.listdir(folder_path):
if file_name.endswith('.xyz'):
file_path = os.path.join(folder_path, file_name)
atoms_list = read(file_path, index=':')
all_atoms.extend(atoms_list)
sorted_configurations = sorted(all_atoms, key=lambda config: config.get_potential_energy())
write(output_file, sorted_configurations)
folder_path = output_foldername
output_filename = input("Enter the output XYZ file name: ")
output_file = output_filename
merge_and_sort_xyz_files(folder_path, output_file)

You can convert this usuing this code written by me

[BianTieyuan]

Thanks! @Amitcuhp