3999396905

Build Type

push

github-actions

Committed by GitHub

Commit Message

[merge before other PRs] ruff updates (#580)

Run Details

960 of 1311 branches covered (73.23%)

Branch coverage included in aggregate %.

15 of 15 new or added lines in 11 files covered. (100.0%)

4034 of 4930 relevant lines covered (81.83%)

3.19 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.17

/mdsuite/project/project.py

"""
MDSuite: A Zincwarecode package.
License
-------
This program and the accompanying materials are made available under the terms
of the Eclipse Public License v2.0 which accompanies this distribution, and is
available at https://www.eclipse.org/legal/epl-v20.html
SPDX-License-Identifier: EPL-2.0
Copyright Contributors to the Zincwarecode Project.
Contact Information
-------------------
email: zincwarecode@gmail.com
github: https://github.com/zincware
web: https://zincwarecode.com/
Citation
--------
If you use this module please cite us with:
Summary
-------.
"""
from __future__ import annotations

import logging
import pathlib
from datetime import datetime
from pathlib import Path
from typing import Dict, Union

from dot4dict import dotdict

import mdsuite.database.scheme as db
import mdsuite.file_io.file_read
from mdsuite.database.project_database import ProjectDatabase
from mdsuite.experiment import Experiment
from mdsuite.experiment.run import RunComputation
from mdsuite.utils import Units
from mdsuite.utils.helpers import NoneType

log = logging.getLogger(__name__)


class Project(ProjectDatabase):
    """Class for the main container of all experiments.

    The Project class acts as the encompassing class for analysis with MDSuite.
    It contains all method required to add and analyze new experiments. These
    experiments may then be compared with one another quickly. The state of the
    class is saved and updated after each operation in order to retain the
    most current state of the analysis.

    .. code-block:: python

        project = mdsuite.Project()
        project.add_experiment(
            name="NaCl",
            timestep=0.002,
            temperature=1400.0,
            units="metal",
            simulation_data="NaCl_gk_i_q.lammpstraj",
            active=False # calculations are only performed on active experiments
            )
        project.activate_experiments("NaCl") # set experiment state to active
        project.run.RadialDistributionFunction(number_of_configurations=500)
        project.disable_experiments("NaCl") # set experiment state to inactive

    Attributes
    ----------
    name : str
            The name of the project
    description : str
            A short description of the project
    storage_path : str
            Where to store the tensor_values and databases. This may not simply
            be the current directory if the databases are expected to be
            quite large.
    experiments : dict
            A dict of class objects. Class objects are instances of the experiment class
            for different experiments.
    """

    def __init__(
        self,
        name: str = None,
        storage_path: Union[str, Path] = "./",
        description: str = None,
    ):
        """Project class constructor.

        The constructor will check to see if the project already exists, if so,
        it will load the state of each of the classes so that they can be used
        again. If the project is new, the constructor will build the necessary
        file structure for the project.

        Parameters
        ----------
        name : str
                The name of the project.
        storage_path : str
                Where to store the tensor_values and databases. This should be
                a place with sufficient storage space for the full analysis.
        """
        super().__init__()
        if name is None:
            self.name = "MDSuite_Project"
        else:
            self.name = name
        self.storage_path = Path(storage_path).as_posix()

        # Properties
        self._experiments = {}

        # Check for project directory, if none exist, create a new one
        self.project_dir = Path(f"{self.storage_path}/{self.name}")

        if self.project_dir.exists():
            self.attach_file_logger()
            log.info("Loading the class state")
            log.info(f"Available experiments are: {self.db_experiments}")
        else:
            self.project_dir.mkdir(parents=True, exist_ok=True)
            self.attach_file_logger()
            log.info(f"Creating new project {self.name}")

        self.build_database()

        # Database Properties
        self.description = description

    def attach_file_logger(self):
        """Attach a file logger for this project."""
        logger = logging.getLogger("mdsuite")
        formatter = logging.Formatter(
            "%(asctime)s %(levelname)s (%(module)s): %(message)s"
        )
        # TODO this will potentially log two mds.Projects into the same file
        #   maybe there are some conditional logging Handlers that can check
        #   project.name, but for now this should be fine.
        channel = logging.FileHandler(self.project_dir / "mdsuite.log")
        channel.setLevel(logging.DEBUG)
        channel.setFormatter(formatter)

        logger.addHandler(channel)

    def __str__(self):
        r"""

        Returns
        -------
        str:
            A list of all available experiments like "1.) Exp01\n2.) Exp02\n3.) Exp03"
        """
        return "\n".join([f"{exp.id}.) {exp.name}" for exp in self.db_experiments])

    def add_experiment(
        self,
        name: str = NoneType,
        timestep: float = None,
        temperature: float = None,
        units: Union[str, Units] = None,
        cluster_mode: bool = None,
        active: bool = True,
        simulation_data: Union[
            str, pathlib.Path, mdsuite.file_io.file_read.FileProcessor, list
        ] = None,  # TODO make this the second argument, (name, data, ...)
    ) -> Experiment:
        """Add an experiment to the project.

        Parameters
        ----------
        active: bool, default = True
                Activate the experiment when added
        cluster_mode : bool
                If true, cluster mode is parsed to the experiment class.
        name : str
                Name to use for the experiment.
        timestep : float
                Timestep used during the simulation.
        temperature : float
                Temperature the simulation was performed at and is to be used
                in calculation.
        units : str
                units used
        simulation_data:
            data that should be added to the experiment.
            see mdsuite.experiment.add_data() for details of the file specification.
            you can also create the experiment with simulation_data == None and add data
            later
        Notes
        ------
        Using custom NoneType to raise a custom ValueError message with useful info.

        Returns
        -------
        Experiment:
            The experiment object that was added to the project

        """
        if name is NoneType:
            raise ValueError(
                "Experiment name can not be empty! "
                "Use None to automatically generate a unique name."
            )

        if name is None:
            name = f"Experiment_{datetime.now().strftime('%Y%m%d-%H%M%S')}"
            # set the experiment name to the current date and time if None is provided

        # Run a query to see if that experiment already exists
        with self.session as ses:
            experiments = (
                ses.query(db.Experiment).filter(db.Experiment.name == name).all()
            )
        if len(experiments) > 0:
            log.info("This experiment already exists")
            self.load_experiments(name)
            return self.experiments[name]

        # If the experiment does not exists, instantiate a new Experiment
        new_experiment = Experiment(
            project=self,
            name=name,
            time_step=timestep,
            temperature=temperature,
            units=units,
            cluster_mode=cluster_mode,
        )

        new_experiment.active = active

        # Update the internal experiment dictionary for self.experiment property
        self._experiments[name] = new_experiment

        if simulation_data is not None:
            self.experiments[name].add_data(simulation_data)

        return self.experiments[name]

    def load_experiments(self, names: Union[str, list]):
        """Alias for activate_experiments."""
        self.activate_experiments(names)

    def activate_experiments(self, names: Union[str, list]):
        """Load experiments, such that they are used for the computations.

        Parameters
        ----------
        names: Name or list of names of experiments that should be instantiated
               and loaded into self.experiments.

        Returns
        -------
        Updates the class state.
        """
        if isinstance(names, str):
            names = [names]

        for name in names:
            self.experiments[name].active = True

    def disable_experiments(self, names: Union[str, list]):
        """Disable experiments.

        Parameters
        ----------
        names: Name or list of names of experiments that should be instantiated
               and loaded into self.experiments
        Returns
        -------

        """
        if isinstance(names, str):
            names = [names]

        for name in names:
            self.experiments[name].active = False

    def add_data(self, data_sets: dict):
        """Add simulation_data to a experiments.

        This is a method so that parallelization is
        possible amongst simulation_data addition to different experiments at the same
        time.

        Parameters
        ----------
        data_sets: dict
            keys: the names of the experiments
            values: str or mdsuite.file_io.file_read.FileProcessor
                refer to mdsuite.experiment.add_data() for an explanation of the file
                specification options
        Returns
        -------
        Updates the experiment classes.
        """
        for key, val in data_sets.items():
            self.experiments[key].add_data(val)

    @property
    def run(self) -> RunComputation:
        """Method to access the available calculators.

        Returns
        -------
        RunComputation:
            class that has all available calculators as properties
        """
        return RunComputation(experiments=list(self.active_experiments.values()))

    @property
    def experiments(self) -> Dict[str, Experiment]:
        """Get a DotDict of instantiated experiments!."""
        with self.session as ses:
            db_experiments = ses.query(db.Experiment).all()

        for exp in db_experiments:
            exp: db.Experiment
            if exp.name not in self._experiments:
                self._experiments[exp.name] = Experiment(project=self, name=exp.name)

        return dotdict(self._experiments)

    @property
    def active_experiments(self) -> Dict[str, Experiment]:
        """Get a DotDict of instantiated experiments that are currently selected!."""
        active_experiment = {
            key: val for key, val in self.experiments.items() if val.active
        }

        return dotdict(active_experiment)

1	"""
2	MDSuite: A Zincwarecode package.
3	License
4	-------
5	This program and the accompanying materials are made available under the terms
6	of the Eclipse Public License v2.0 which accompanies this distribution, and is
7	available at https://www.eclipse.org/legal/epl-v20.html
8	SPDX-License-Identifier: EPL-2.0
9	Copyright Contributors to the Zincwarecode Project.
10	Contact Information
11	-------------------
12	email: zincwarecode@gmail.com
13	github: https://github.com/zincware
14	web: https://zincwarecode.com/
15	Citation
16	--------
17	If you use this module please cite us with:
18	Summary
19	-------.
20	"""
21	from __future__ import annotations	4✔
22
23	import logging	4✔
24	import pathlib	4✔
25	from datetime import datetime	4✔
26	from pathlib import Path	4✔
27	from typing import Dict, Union	4✔
28
29	from dot4dict import dotdict	4✔
30
31	import mdsuite.database.scheme as db	4✔
32	import mdsuite.file_io.file_read	4✔
33	from mdsuite.database.project_database import ProjectDatabase	4✔
34	from mdsuite.experiment import Experiment	4✔
35	from mdsuite.experiment.run import RunComputation	4✔
36	from mdsuite.utils import Units	4✔
37	from mdsuite.utils.helpers import NoneType	4✔
38
39	log = logging.getLogger(__name__)	4✔
40
41
42	class Project(ProjectDatabase):	4✔
43	"""Class for the main container of all experiments.
44
45	The Project class acts as the encompassing class for analysis with MDSuite.
46	It contains all method required to add and analyze new experiments. These
47	experiments may then be compared with one another quickly. The state of the
48	class is saved and updated after each operation in order to retain the
49	most current state of the analysis.
50
51	.. code-block:: python
52
53	project = mdsuite.Project()
54	project.add_experiment(
55	name="NaCl",
56	timestep=0.002,
57	temperature=1400.0,
58	units="metal",
59	simulation_data="NaCl_gk_i_q.lammpstraj",
60	active=False # calculations are only performed on active experiments
61	)
62	project.activate_experiments("NaCl") # set experiment state to active
63	project.run.RadialDistributionFunction(number_of_configurations=500)
64	project.disable_experiments("NaCl") # set experiment state to inactive
65
66	Attributes
67	----------
68	name : str
69	The name of the project
70	description : str
71	A short description of the project
72	storage_path : str
73	Where to store the tensor_values and databases. This may not simply
74	be the current directory if the databases are expected to be
75	quite large.
76	experiments : dict
77	A dict of class objects. Class objects are instances of the experiment class
78	for different experiments.
79	"""
80
81	def __init__(	4✔
82	self,
83	name: str = None,
84	storage_path: Union[str, Path] = "./",
85	description: str = None,
86	):
87	"""Project class constructor.
88
89	The constructor will check to see if the project already exists, if so,
90	it will load the state of each of the classes so that they can be used
91	again. If the project is new, the constructor will build the necessary
92	file structure for the project.
93
94	Parameters
95	----------
96	name : str
97	The name of the project.
98	storage_path : str
99	Where to store the tensor_values and databases. This should be
100	a place with sufficient storage space for the full analysis.
101	"""
102	super().__init__()	4✔
103	if name is None:	4✔
104	self.name = "MDSuite_Project"	4✔
105	else:
106	self.name = name	4✔
107	self.storage_path = Path(storage_path).as_posix()	4✔
108
109	# Properties
110	self._experiments = {}	4✔
111
112	# Check for project directory, if none exist, create a new one
113	self.project_dir = Path(f"{self.storage_path}/{self.name}")	4✔
114
115	if self.project_dir.exists():	4✔
116	self.attach_file_logger()	4✔
117	log.info("Loading the class state")	4✔
118	log.info(f"Available experiments are: {self.db_experiments}")	4✔
119	else:
120	self.project_dir.mkdir(parents=True, exist_ok=True)	4✔
121	self.attach_file_logger()	4✔
122	log.info(f"Creating new project {self.name}")	4✔
123
124	self.build_database()	4✔
125
126	# Database Properties
127	self.description = description	4✔
128
129	def attach_file_logger(self):	4✔
130	"""Attach a file logger for this project."""
131	logger = logging.getLogger("mdsuite")	4✔
132	formatter = logging.Formatter(	4✔
133	"%(asctime)s %(levelname)s (%(module)s): %(message)s"
134	)
135	# TODO this will potentially log two mds.Projects into the same file
136	# maybe there are some conditional logging Handlers that can check
137	# project.name, but for now this should be fine.
138	channel = logging.FileHandler(self.project_dir / "mdsuite.log")	4✔
139	channel.setLevel(logging.DEBUG)	4✔
140	channel.setFormatter(formatter)	4✔
141
142	logger.addHandler(channel)	4✔
143
144	def __str__(self):	4✔
145	r"""
146
147	Returns
148	-------
149	str:
150	A list of all available experiments like "1.) Exp01\n2.) Exp02\n3.) Exp03"
151	"""
152	return "\n".join([f"{exp.id}.) {exp.name}" for exp in self.db_experiments])	×
153
154	def add_experiment(	4✔
155	self,
156	name: str = NoneType,
157	timestep: float = None,
158	temperature: float = None,
159	units: Union[str, Units] = None,
160	cluster_mode: bool = None,
161	active: bool = True,
162	simulation_data: Union[
163	str, pathlib.Path, mdsuite.file_io.file_read.FileProcessor, list
164	] = None, # TODO make this the second argument, (name, data, ...)
165	) -> Experiment:
166	"""Add an experiment to the project.
167
168	Parameters
169	----------
170	active: bool, default = True
171	Activate the experiment when added
172	cluster_mode : bool
173	If true, cluster mode is parsed to the experiment class.
174	name : str
175	Name to use for the experiment.
176	timestep : float
177	Timestep used during the simulation.
178	temperature : float
179	Temperature the simulation was performed at and is to be used
180	in calculation.
181	units : str
182	units used
183	simulation_data:
184	data that should be added to the experiment.
185	see mdsuite.experiment.add_data() for details of the file specification.
186	you can also create the experiment with simulation_data == None and add data
187	later
188	Notes
189	------
190	Using custom NoneType to raise a custom ValueError message with useful info.
191
192	Returns
193	-------
194	Experiment:
195	The experiment object that was added to the project
196
197	"""
198	if name is NoneType:	4!
199	raise ValueError(	×
200	"Experiment name can not be empty! "
201	"Use None to automatically generate a unique name."
202	)
203
204	if name is None:	4!
205	name = f"Experiment_{datetime.now().strftime('%Y%m%d-%H%M%S')}"	×
206	# set the experiment name to the current date and time if None is provided
207
208	# Run a query to see if that experiment already exists
209	with self.session as ses:	4✔
210	experiments = (	4✔
211	ses.query(db.Experiment).filter(db.Experiment.name == name).all()
212	)
213	if len(experiments) > 0:	4✔
214	log.info("This experiment already exists")	4✔
215	self.load_experiments(name)	4✔
216	return self.experiments[name]	4✔
217
218	# If the experiment does not exists, instantiate a new Experiment
219	new_experiment = Experiment(	4✔
220	project=self,
221	name=name,
222	time_step=timestep,
223	temperature=temperature,
224	units=units,
225	cluster_mode=cluster_mode,
226	)
227
228	new_experiment.active = active	4✔
229
230	# Update the internal experiment dictionary for self.experiment property
231	self._experiments[name] = new_experiment	4✔
232
233	if simulation_data is not None:	4✔
234	self.experiments[name].add_data(simulation_data)	4✔
235
236	return self.experiments[name]	4✔
237
238	def load_experiments(self, names: Union[str, list]):	4✔
239	"""Alias for activate_experiments."""
240	self.activate_experiments(names)	4✔
241
242	def activate_experiments(self, names: Union[str, list]):	4✔
243	"""Load experiments, such that they are used for the computations.
244
245	Parameters
246	----------
247	names: Name or list of names of experiments that should be instantiated
248	and loaded into self.experiments.
249
250	Returns
251	-------
252	Updates the class state.
253	"""
254	if isinstance(names, str):	4✔
255	names = [names]	4✔
256
257	for name in names:	4✔
258	self.experiments[name].active = True	4✔
259
260	def disable_experiments(self, names: Union[str, list]):	4✔
261	"""Disable experiments.
262
263	Parameters
264	----------
265	names: Name or list of names of experiments that should be instantiated
266	and loaded into self.experiments
267	Returns
268	-------
269
270	"""
271	if isinstance(names, str):	×
272	names = [names]	×
273
274	for name in names:	×
275	self.experiments[name].active = False	×
276
277	def add_data(self, data_sets: dict):	4✔
278	"""Add simulation_data to a experiments.
279
280	This is a method so that parallelization is
281	possible amongst simulation_data addition to different experiments at the same
282	time.
283
284	Parameters
285	----------
286	data_sets: dict
287	keys: the names of the experiments
288	values: str or mdsuite.file_io.file_read.FileProcessor
289	refer to mdsuite.experiment.add_data() for an explanation of the file
290	specification options
291	Returns
292	-------
293	Updates the experiment classes.
294	"""
295	for key, val in data_sets.items():	×
296	self.experiments[key].add_data(val)	×
297
298	@property	4✔
299	def run(self) -> RunComputation:	3✔
300	"""Method to access the available calculators.
301
302	Returns
303	-------
304	RunComputation:
305	class that has all available calculators as properties
306	"""
307	return RunComputation(experiments=list(self.active_experiments.values()))	4✔
308
309	@property	4✔
310	def experiments(self) -> Dict[str, Experiment]:	3✔
311	"""Get a DotDict of instantiated experiments!."""
312	with self.session as ses:	4✔
313	db_experiments = ses.query(db.Experiment).all()	4✔
314
315	for exp in db_experiments:	4✔
316	exp: db.Experiment
317	if exp.name not in self._experiments:	4✔
318	self._experiments[exp.name] = Experiment(project=self, name=exp.name)	4✔
319
320	return dotdict(self._experiments)	4✔
321
322	@property	4✔
323	def active_experiments(self) -> Dict[str, Experiment]:	3✔
324	"""Get a DotDict of instantiated experiments that are currently selected!."""
325	active_experiment = {	4✔
326	key: val for key, val in self.experiments.items() if val.active
327	}
328
329	return dotdict(active_experiment)	4✔

zincware / MDSuite / 3999396905

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous