14972158685

Committed 12 May 2025 12:25PM UTC coverage: 40.957% (+0.4%) from 40.52%

Build # 14972158685

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #88 from CCPBioSim/85-oop-refactor

Object-oriented programming Refactor for CodeEntropy

Run Details

168 of 563 new or added lines in 4 files covered. (29.84%)

274 of 669 relevant lines covered (40.96%)

1.23 hits per line

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

54.46

/CodeEntropy/run.py

import logging
import os
import pickle

import MDAnalysis as mda
from MDAnalysis.analysis.base import AnalysisFromFunction
from MDAnalysis.coordinates.memory import MemoryReader

from CodeEntropy.config.arg_config_manager import ConfigManager
from CodeEntropy.config.data_logger import DataLogger
from CodeEntropy.config.logging_config import LoggingConfig
from CodeEntropy.entropy import EntropyManager
from CodeEntropy.levels import LevelManager

logger = logging.getLogger(__name__)


class RunManager:
    """
    Handles the setup and execution of entropy analysis runs, including configuration
    loading, logging, and access to physical constants used in calculations.
    """

    def __init__(self, folder):
        """
        Initializes the RunManager with the working folder and sets up configuration,
        data logging, and logging systems. Also defines physical constants used in
        entropy calculations.
        """
        self.folder = folder
        self._config_manager = ConfigManager()
        self._data_logger = DataLogger()
        self._logging_config = LoggingConfig(folder)
        self._N_AVOGADRO = 6.0221415e23
        self._DEF_TEMPER = 298

    @property
    def N_AVOGADRO(self):
        """Returns Avogadro's number used in entropy calculations."""
        return self._N_AVOGADRO

    @property
    def DEF_TEMPER(self):
        """Returns the default temperature (in Kelvin) used in the analysis."""
        return self._DEF_TEMPER

    @staticmethod
    def create_job_folder():
        """
        Create a new job folder with an incremented job number based on existing
        folders.
        """
        # Get the current working directory
        current_dir = os.getcwd()

        # Get a list of existing folders that start with "job"
        existing_folders = [f for f in os.listdir(current_dir) if f.startswith("job")]

        # Extract numbers from existing folder names
        job_numbers = []
        for folder in existing_folders:
            try:
                # Assuming folder names are in the format "jobXXX"
                job_number = int(folder[3:])  # Get the number part after "job"
                job_numbers.append(job_number)
            except ValueError:
                continue  # Ignore any folder names that don't follow the pattern

        # If no folders exist, start with job001
        if not job_numbers:
            next_job_number = 1
        else:
            next_job_number = max(job_numbers) + 1

        # Create the new job folder name
        new_job_folder = f"job{next_job_number:03d}"

        # Create the full path to the new folder
        new_folder_path = os.path.join(current_dir, new_job_folder)

        # Create the directory
        os.makedirs(new_folder_path, exist_ok=True)

        # Return the path of the newly created folder
        return new_folder_path

    def run_entropy_workflow(self):
        """
        Runs the entropy analysis workflow by setting up logging, loading configuration
        files, parsing arguments, and executing the analysis for each configured run.
        Initializes the MDAnalysis Universe and supporting managers, and logs all
        relevant inputs and commands.
        """
        try:
            logger = self._logging_config.setup_logging()

            config = self._config_manager.load_config("config.yaml")
            if config is None:
                raise ValueError(
                    "No configuration file found, and no CLI arguments were provided."
                )

            parser = self._config_manager.setup_argparse()
            args, _ = parser.parse_known_args()
            args.output_file = os.path.join(self.folder, args.output_file)

            for run_name, run_config in config.items():
                if not isinstance(run_config, dict):
                    logger.warning(
                        f"Run configuration for {run_name} is not a dictionary."
                    )
                    continue

                args = self._config_manager.merge_configs(args, run_config)

                log_level = logging.DEBUG if args.verbose else logging.INFO
                self._logging_config.update_logging_level(log_level)

                command = " ".join(os.sys.argv)
                logging.getLogger("commands").info(command)

                if not getattr(args, "top_traj_file", None):
                    raise ValueError("Missing 'top_traj_file' argument.")
                if not getattr(args, "selection_string", None):
                    raise ValueError("Missing 'selection_string' argument.")

                # Log all inputs for the current run
                logger.info(f"All input for {run_name}")
                for arg in vars(args):
                    logger.info(f" {arg}: {getattr(args, arg) or ''}")

                # Load MDAnalysis Universe
                tprfile = args.top_traj_file[0]
                trrfile = args.top_traj_file[1:]
                logger.debug(f"Loading Universe with {tprfile} and {trrfile}")
                u = mda.Universe(tprfile, trrfile)

                # Create LevelManager instance
                level_manager = LevelManager()

                # Inject all dependencies into EntropyManager
                entropy_manager = EntropyManager(
                    run_manager=self,
                    args=args,
                    universe=u,
                    data_logger=self._data_logger,
                    level_manager=level_manager,
                )

                entropy_manager.execute()

        except Exception as e:
            logger.error(f"RunManager encountered an error: {e}", exc_info=True)
            raise

    def new_U_select_frame(self, u, start=None, end=None, step=1):
        """Create a reduced universe by dropping frames according to user selection

        Parameters
        ----------
        u : MDAnalyse.Universe
            A Universe object will all topology, dihedrals,coordinates and force
            information
        start : int or None, Optional, default: None
            Frame id to start analysis. Default None will start from frame 0
        end : int or None, Optional, default: None
            Frame id to end analysis. Default None will end at last frame
        step : int, Optional, default: 1
            Steps between frame.

        Returns
        -------
            u2 : MDAnalysis.Universe
                reduced universe
        """
        if start is None:
            start = 0
        if end is None:
            end = len(u.trajectory)
        select_atom = u.select_atoms("all", updating=True)
        coordinates = (
            AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
            .run()
            .results["timeseries"][start:end:step]
        )
        forces = (
            AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
            .run()
            .results["timeseries"][start:end:step]
        )
        dimensions = (
            AnalysisFromFunction(lambda ag: ag.dimensions.copy(), select_atom)
            .run()
            .results["timeseries"][start:end:step]
        )
        u2 = mda.Merge(select_atom)
        u2.load_new(
            coordinates, format=MemoryReader, forces=forces, dimensions=dimensions
        )
        logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")
        return u2

    def new_U_select_atom(self, u, select_string="all"):
        """Create a reduced universe by dropping atoms according to user selection

        Parameters
        ----------
        u : MDAnalyse.Universe
            A Universe object will all topology, dihedrals,coordinates and force
            information
        select_string : str, Optional, default: 'all'
            MDAnalysis.select_atoms selection string.

        Returns
        -------
            u2 : MDAnalysis.Universe
                reduced universe

        """
        select_atom = u.select_atoms(select_string, updating=True)
        coordinates = (
            AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
            .run()
            .results["timeseries"]
        )
        forces = (
            AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
            .run()
            .results["timeseries"]
        )
        dimensions = (
            AnalysisFromFunction(lambda ag: ag.dimensions.copy(), select_atom)
            .run()
            .results["timeseries"]
        )
        u2 = mda.Merge(select_atom)
        u2.load_new(
            coordinates, format=MemoryReader, forces=forces, dimensions=dimensions
        )
        logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")
        return u2

    def write_universe(self, u, name="default"):
        """Write a universe to working directories as pickle

        Parameters
        ----------
        u : MDAnalyse.Universe
            A Universe object will all topology, dihedrals,coordinates and force
            information
        name : str, Optional. default: 'default'
            The name of file with sub file name .pkl

        Returns
        -------
            name : str
                filename of saved universe
        """
        filename = f"{name}.pkl"
        pickle.dump(u, open(filename, "wb"))
        return name

    def read_universe(self, path):
        """read a universe to working directories as pickle

        Parameters
        ----------
        path : str
            The path to file.

        Returns
        -------
            u : MDAnalysis.Universe
                A Universe object will all topology, dihedrals,coordinates and force
                information.
        """
        u = pickle.load(open(path, "rb"))
        return u

    def change_lambda_units(self, arg_lambdas):
        """Unit of lambdas : kJ2 mol-2 A-2 amu-1
        change units of lambda to J/s2"""
        # return arg_lambdas * N_AVOGADRO * N_AVOGADRO * AMU2KG * 1e-26
        return arg_lambdas * 1e29 / self.N_AVOGADRO

    def get_KT2J(self, arg_temper):
        """A temperature dependent KT to Joule conversion"""
        return 4.11e-21 * arg_temper / self.DEF_TEMPER

1	import logging	3✔
2	import os	3✔
3	import pickle	3✔
4
5	import MDAnalysis as mda	3✔
6	from MDAnalysis.analysis.base import AnalysisFromFunction	3✔
7	from MDAnalysis.coordinates.memory import MemoryReader	3✔
8
9	from CodeEntropy.config.arg_config_manager import ConfigManager	3✔
10	from CodeEntropy.config.data_logger import DataLogger	3✔
11	from CodeEntropy.config.logging_config import LoggingConfig	3✔
12	from CodeEntropy.entropy import EntropyManager	3✔
13	from CodeEntropy.levels import LevelManager	3✔
14
15	logger = logging.getLogger(__name__)	3✔
16
17
18	class RunManager:	3✔
19	"""
20	Handles the setup and execution of entropy analysis runs, including configuration
21	loading, logging, and access to physical constants used in calculations.
22	"""
23
24	def __init__(self, folder):	3✔
25	"""
26	Initializes the RunManager with the working folder and sets up configuration,
27	data logging, and logging systems. Also defines physical constants used in
28	entropy calculations.
29	"""
30	self.folder = folder	3✔
31	self._config_manager = ConfigManager()	3✔
32	self._data_logger = DataLogger()	3✔
33	self._logging_config = LoggingConfig(folder)	3✔
34	self._N_AVOGADRO = 6.0221415e23	3✔
35	self._DEF_TEMPER = 298	3✔
36
37	@property	3✔
38	def N_AVOGADRO(self):	3✔
39	"""Returns Avogadro's number used in entropy calculations."""
40	return self._N_AVOGADRO	3✔
41
42	@property	3✔
43	def DEF_TEMPER(self):	3✔
44	"""Returns the default temperature (in Kelvin) used in the analysis."""
45	return self._DEF_TEMPER	3✔
46
47	@staticmethod	3✔
48	def create_job_folder():	3✔
49	"""
50	Create a new job folder with an incremented job number based on existing
51	folders.
52	"""
53	# Get the current working directory
54	current_dir = os.getcwd()	3✔
55
56	# Get a list of existing folders that start with "job"
57	existing_folders = [f for f in os.listdir(current_dir) if f.startswith("job")]	3✔
58
59	# Extract numbers from existing folder names
60	job_numbers = []	3✔
61	for folder in existing_folders:	3✔
62	try:	3✔
63	# Assuming folder names are in the format "jobXXX"
64	job_number = int(folder[3:]) # Get the number part after "job"	3✔
65	job_numbers.append(job_number)	3✔
66	except ValueError:	3✔
67	continue # Ignore any folder names that don't follow the pattern	3✔
68
69	# If no folders exist, start with job001
70	if not job_numbers:	3✔
71	next_job_number = 1	3✔
72	else:
73	next_job_number = max(job_numbers) + 1	3✔
74
75	# Create the new job folder name
76	new_job_folder = f"job{next_job_number:03d}"	3✔
77
78	# Create the full path to the new folder
79	new_folder_path = os.path.join(current_dir, new_job_folder)	3✔
80
81	# Create the directory
82	os.makedirs(new_folder_path, exist_ok=True)	3✔
83
84	# Return the path of the newly created folder
85	return new_folder_path	3✔
86
87	def run_entropy_workflow(self):	3✔
88	"""
89	Runs the entropy analysis workflow by setting up logging, loading configuration
90	files, parsing arguments, and executing the analysis for each configured run.
91	Initializes the MDAnalysis Universe and supporting managers, and logs all
92	relevant inputs and commands.
93	"""
94	try:	3✔
95	logger = self._logging_config.setup_logging()	3✔
96
97	config = self._config_manager.load_config("config.yaml")	3✔
98	if config is None:	3✔
99	raise ValueError(	3✔
100	"No configuration file found, and no CLI arguments were provided."
101	)
102
NEW 103	parser = self._config_manager.setup_argparse()	×
NEW 104	args, _ = parser.parse_known_args()	×
NEW 105	args.output_file = os.path.join(self.folder, args.output_file)	×
106
NEW 107	for run_name, run_config in config.items():	×
NEW 108	if not isinstance(run_config, dict):	×
NEW 109	logger.warning(	×
110	f"Run configuration for {run_name} is not a dictionary."
111	)
NEW 112	continue	×
113
NEW 114	args = self._config_manager.merge_configs(args, run_config)	×
115
NEW 116	log_level = logging.DEBUG if args.verbose else logging.INFO	×
NEW 117	self._logging_config.update_logging_level(log_level)	×
118
NEW 119	command = " ".join(os.sys.argv)	×
NEW 120	logging.getLogger("commands").info(command)	×
121
NEW 122	if not getattr(args, "top_traj_file", None):	×
NEW 123	raise ValueError("Missing 'top_traj_file' argument.")	×
NEW 124	if not getattr(args, "selection_string", None):	×
NEW 125	raise ValueError("Missing 'selection_string' argument.")	×
126
127	# Log all inputs for the current run
NEW 128	logger.info(f"All input for {run_name}")	×
NEW 129	for arg in vars(args):	×
NEW 130	logger.info(f" {arg}: {getattr(args, arg) or ''}")	×
131
132	# Load MDAnalysis Universe
NEW 133	tprfile = args.top_traj_file[0]	×
NEW 134	trrfile = args.top_traj_file[1:]	×
NEW 135	logger.debug(f"Loading Universe with {tprfile} and {trrfile}")	×
NEW 136	u = mda.Universe(tprfile, trrfile)	×
137
138	# Create LevelManager instance
NEW 139	level_manager = LevelManager()	×
140
141	# Inject all dependencies into EntropyManager
NEW 142	entropy_manager = EntropyManager(	×
143	run_manager=self,
144	args=args,
145	universe=u,
146	data_logger=self._data_logger,
147	level_manager=level_manager,
148	)
149
NEW 150	entropy_manager.execute()	×
151
152	except Exception as e:	3✔
153	logger.error(f"RunManager encountered an error: {e}", exc_info=True)	3✔
154	raise	3✔
155
156	def new_U_select_frame(self, u, start=None, end=None, step=1):	3✔
157	"""Create a reduced universe by dropping frames according to user selection
158
159	Parameters
160	----------
161	u : MDAnalyse.Universe
162	A Universe object will all topology, dihedrals,coordinates and force
163	information
164	start : int or None, Optional, default: None
165	Frame id to start analysis. Default None will start from frame 0
166	end : int or None, Optional, default: None
167	Frame id to end analysis. Default None will end at last frame
168	step : int, Optional, default: 1
169	Steps between frame.
170
171	Returns
172	-------
173	u2 : MDAnalysis.Universe
174	reduced universe
175	"""
NEW 176	if start is None:	×
NEW 177	start = 0	×
NEW 178	if end is None:	×
NEW 179	end = len(u.trajectory)	×
NEW 180	select_atom = u.select_atoms("all", updating=True)	×
NEW 181	coordinates = (	×
182	AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
183	.run()
184	.results["timeseries"][start:end:step]
185	)
NEW 186	forces = (	×
187	AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
188	.run()
189	.results["timeseries"][start:end:step]
190	)
NEW 191	dimensions = (	×
192	AnalysisFromFunction(lambda ag: ag.dimensions.copy(), select_atom)
193	.run()
194	.results["timeseries"][start:end:step]
195	)
NEW 196	u2 = mda.Merge(select_atom)	×
NEW 197	u2.load_new(	×
198	coordinates, format=MemoryReader, forces=forces, dimensions=dimensions
199	)
NEW 200	logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")	×
NEW 201	return u2	×
202
203	def new_U_select_atom(self, u, select_string="all"):	3✔
204	"""Create a reduced universe by dropping atoms according to user selection
205
206	Parameters
207	----------
208	u : MDAnalyse.Universe
209	A Universe object will all topology, dihedrals,coordinates and force
210	information
211	select_string : str, Optional, default: 'all'
212	MDAnalysis.select_atoms selection string.
213
214	Returns
215	-------
216	u2 : MDAnalysis.Universe
217	reduced universe
218
219	"""
NEW 220	select_atom = u.select_atoms(select_string, updating=True)	×
NEW 221	coordinates = (	×
222	AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
223	.run()
224	.results["timeseries"]
225	)
NEW 226	forces = (	×
227	AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
228	.run()
229	.results["timeseries"]
230	)
NEW 231	dimensions = (	×
232	AnalysisFromFunction(lambda ag: ag.dimensions.copy(), select_atom)
233	.run()
234	.results["timeseries"]
235	)
NEW 236	u2 = mda.Merge(select_atom)	×
NEW 237	u2.load_new(	×
238	coordinates, format=MemoryReader, forces=forces, dimensions=dimensions
239	)
NEW 240	logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")	×
NEW 241	return u2	×
242
243	def write_universe(self, u, name="default"):	3✔
244	"""Write a universe to working directories as pickle
245
246	Parameters
247	----------
248	u : MDAnalyse.Universe
249	A Universe object will all topology, dihedrals,coordinates and force
250	information
251	name : str, Optional. default: 'default'
252	The name of file with sub file name .pkl
253
254	Returns
255	-------
256	name : str
257	filename of saved universe
258	"""
NEW 259	filename = f"{name}.pkl"	×
NEW 260	pickle.dump(u, open(filename, "wb"))	×
NEW 261	return name	×
262
263	def read_universe(self, path):	3✔
264	"""read a universe to working directories as pickle
265
266	Parameters
267	----------
268	path : str
269	The path to file.
270
271	Returns
272	-------
273	u : MDAnalysis.Universe
274	A Universe object will all topology, dihedrals,coordinates and force
275	information.
276	"""
NEW 277	u = pickle.load(open(path, "rb"))	×
NEW 278	return u	×
279
280	def change_lambda_units(self, arg_lambdas):	3✔
281	"""Unit of lambdas : kJ2 mol-2 A-2 amu-1
282	change units of lambda to J/s2"""
283	# return arg_lambdas * N_AVOGADRO * N_AVOGADRO * AMU2KG * 1e-26
284	return arg_lambdas * 1e29 / self.N_AVOGADRO	3✔
285
286	def get_KT2J(self, arg_temper):	3✔
287	"""A temperature dependent KT to Joule conversion"""
288	return 4.11e-21 * arg_temper / self.DEF_TEMPER	3✔

CCPBioSim / CodeEntropy / 14972158685

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