Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self):
if self.project.load('trial') is not None:
self.project.remove_job("trial")
self.job = self.project.create_job("Yaff", "trial")
if self.project.load('yaff_complete') is not None:
self.project.remove_job("yaff_complete")
self.job_complete = Yaff(
project=ProjectHDFio(project=self.project, file_name="yaff_complete"),
job_name="yaff_complete",
)
def setUpClass(cls):
cls.execution_path = os.path.dirname(os.path.abspath(__file__))
cls.project = Project(os.path.join(cls.execution_path, "test_quickff"))
cls.job = cls.project.create_job("QuickFF", "trial")
cls.job_complete = QuickFF(
project=ProjectHDFio(project=cls.project, file_name="quickff_complete"),
job_name="quickff_complete",
)
cls.pl = InputList([
{"foo": "bar"},
2,
42,
{"next":
[0,
{"depth": 23}
]
}
], table_name = "input")
cls.pl["tail"] = InputList([2,4,8])
file_location = os.path.dirname(os.path.abspath(__file__))
pr = Project(file_location)
cls.file_name = os.path.join(file_location, "input.h5")
cls.hdf = ProjectHDFio(project=pr, file_name=cls.file_name,
h5_path="/test", mode="a")
def test_to_object(self):
filename = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"../../static/atomistics",
)
abs_filename = os.path.abspath(filename)
hdf_obj = ProjectHDFio(
project=Project(abs_filename),
file_name="test.h5"
)
pos, cell = generate_fcc_lattice()
basis_store = Atoms(symbols="Al", positions=pos, cell=cell)
basis_store.set_repeat([2, 2, 2])
basis_store.to_hdf(hdf_obj, "simple_structure")
basis = hdf_obj["simple_structure"].to_object()
self.assertEqual(len(basis), 8)
self.assertEqual(basis.get_majority_species()["symbol"], "Al")
self.assertEqual(basis.get_spacegroup()["Number"], 225)
db_entry = db.get_item_by_id(job_id)
if db_entry is None:
raise ValueError("job ID {0} does not exist!".format(job_id))
job_name = db_entry["job"]
h5_path = None
sub_job = db_entry["subjob"]
if sub_job is not None:
if len(sub_job.strip()) > 0:
h5_path = '/'.join(sub_job.split('/')[:-1])
hdf5_file = sub_job.split('/')[1] + '.h5'
gp = GenericPath(root_path=db_entry["projectpath"], project_path=db_entry["project"])
hdf_project = ProjectHDFio(project=Project(path=gp, user=user), file_name=hdf5_file, h5_path=h5_path, mode="r")
super(JobPath, self).__init__(hdf_project, job_name)
self.__name__ = db_entry["hamilton"]
self.__version__ = db_entry["hamversion"]
if 'id' in db_entry:
self._job_id = db_entry["id"]
self._status = db_entry["status"]
self._master_id = db_entry["masterid"]
self._parent_id = db_entry["parentid"]
convert_to_object (bool): convert the object to an pyiron object or only access the HDF5 file - default=True
accessing only the HDF5 file is about an order of magnitude faster, but only
provides limited functionality. Compare the GenericJob object to JobCore object.
Returns:
Project, GenericJob, JobCore, dict, list, float: basically any kind of item inside the project.
"""
if item == "..":
return self.parent_group
if item in self.list_nodes():
if self._inspect_mode or not convert_to_object:
return self.inspect(item)
return self.load(item)
if item in self.list_files(extension="h5"):
file_name = posixpath.join(self.path, "{}.h5".format(item))
return ProjectHDFio(project=self, file_name=file_name)
if item in self.list_files():
file_name = posixpath.join(self.path, "{}".format(item))
with open(file_name) as f:
return f.readlines()
if item in self.list_dirs():
with self.open(item) as new_item:
return new_item.copy()
raise ValueError("Unknown item: {}".format(item))
convert_to_object (bool): convert the object to an pyiron object or only access the HDF5 file - default=True
accessing only the HDF5 file is about an order of magnitude faster, but only
provides limited functionality. Compare the GenericJob object to JobCore object.
Returns:
Project, GenericJob, JobCore, dict, list, float: basically any kind of item inside the project.
"""
if item == "..":
return self.parent_group
if item in self.list_nodes():
if self._inspect_mode or not convert_to_object:
return self.inspect(item)
return self.load(item)
if item in self.list_files(extension="h5"):
file_name = posixpath.join(self.path, "{}.h5".format(item))
return ProjectHDFio(project=self, file_name=file_name)
if item in self.list_files():
file_name = posixpath.join(self.path, "{}".format(item))
with open(file_name) as f:
return f.readlines()
if item in self.list_dirs():
with self.open(item) as new_item:
return new_item.copy()
raise ValueError("Unknown item: {}".format(item))
def copy(self):
"""
Copy the ProjectHDFio object - copying just the Python object but maintaining the same pyiron path
Returns:
ProjectHDFio: copy of the ProjectHDFio object
"""
new_h5 = ProjectHDFio(project=self._project, file_name=self._file_name, h5_path=self._h5_path)
new_h5._filter = self._filter
return new_h5
- 'SerialMaster': series of jobs run in serial
- 'ParallelMaster': series of jobs run in parallel
- 'ScriptJob': Python script or jupyter notebook job container
- 'ListMaster': list of jobs
Args:
job_type (str): job type can be ['ExampleJob', 'SerialMaster', 'ParallelMaster', 'ScriptJob', 'ListMaster']
job_name (str): name of the job
Returns:
GenericJob: job object depending on the job_type selected
"""
job_name = job_name.replace(".", "_")
job = JobType(
job_type,
project=ProjectHDFio(project=self.copy(), file_name=job_name),
job_name=job_name,
job_class_dict=self.job_type.job_class_dict,
)
if self.user is not None:
job.user = self.user
return job