19675730718

Committed 25 Nov 2025 03:56PM UTC coverage: 90.133% (-9.9%) from 100.0%

Build # 19675730718

Build Type

Pull #200

github

Committed by

web-flow

Commit Message

Merge 444701d78 into 51f6db64b

Pull Request Pull Request #200: Update Averaging over Groups of Molecules for Conformational Entropy

Coverage Stats

99 of 210 new or added lines in 5 files covered. (47.14%)

1014 of 1125 relevant lines covered (90.13%)

0.9 hits per line

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

99.35

/CodeEntropy/run.py

import logging
import os
import pickle

import MDAnalysis as mda
import requests
import yaml
from art import text2art
from rich.align import Align
from rich.console import Group
from rich.padding import Padding
from rich.panel import Panel
from rich.rule import Rule
from rich.table import Table
from rich.text import Text

from CodeEntropy.config.arg_config_manager import ConfigManager
from CodeEntropy.config.data_logger import DataLogger
from CodeEntropy.config.logging_config import LoggingConfig
from CodeEntropy.dihedral_tools import DihedralAnalysis
from CodeEntropy.entropy import EntropyManager
from CodeEntropy.group_molecules import GroupMolecules
from CodeEntropy.levels import LevelManager
from CodeEntropy.mda_universe_operations import UniverseOperations

logger = logging.getLogger(__name__)
console = LoggingConfig.get_console()


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 load_citation_data(self):
        """
        Load CITATION.cff from GitHub into memory.
        Return empty dict if offline.
        """
        url = (
            "https://raw.githubusercontent.com/CCPBioSim/"
            "CodeEntropy/refs/heads/main/CITATION.cff"
        )
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            return yaml.safe_load(response.text)
        except requests.exceptions.RequestException:
            return None

    def show_splash(self):
        """Render splash screen with optional citation metadata."""
        citation = self.load_citation_data()

        if citation:
            # ASCII Title
            ascii_title = text2art(citation.get("title", "CodeEntropy"))
            ascii_render = Align.center(Text(ascii_title, style="bold white"))

            # Metadata
            version = citation.get("version", "?")
            release_date = citation.get("date-released", "?")
            url = citation.get("url", citation.get("repository-code", ""))

            version_text = Align.center(
                Text(f"Version {version} | Released {release_date}", style="green")
            )
            url_text = Align.center(Text(url, style="blue underline"))

            # Description block
            abstract = citation.get("abstract", "No description available.")
            description_title = Align.center(
                Text("Description", style="bold magenta underline")
            )
            description_body = Align.center(
                Padding(Text(abstract, style="white", justify="left"), (0, 4))
            )

            # Contributors table
            contributors_title = Align.center(
                Text("Contributors", style="bold magenta underline")
            )

            author_table = Table(
                show_header=True, header_style="bold yellow", box=None, pad_edge=False
            )
            author_table.add_column("Name", style="bold", justify="center")
            author_table.add_column("Affiliation", justify="center")

            for author in citation.get("authors", []):
                name = (
                    f"{author.get('given-names', '')} {author.get('family-names', '')}"
                ).strip()
                affiliation = author.get("affiliation", "")
                author_table.add_row(name, affiliation)

            contributors_table = Align.center(Padding(author_table, (0, 4)))

            # Full layout
            splash_content = Group(
                ascii_render,
                Rule(style="cyan"),
                version_text,
                url_text,
                Text(),
                description_title,
                description_body,
                Text(),
                contributors_title,
                contributors_table,
            )
        else:
            # ASCII Title
            ascii_title = text2art("CodeEntropy")
            ascii_render = Align.center(Text(ascii_title, style="bold white"))

            splash_content = Group(
                ascii_render,
            )

        splash_panel = Panel(
            splash_content,
            title="[bold bright_cyan]Welcome to CodeEntropy",
            title_align="center",
            border_style="bright_cyan",
            padding=(1, 4),
            expand=True,
        )

        console.print(splash_panel)

    def print_args_table(self, args):
        table = Table(title="Run Configuration", expand=True)

        table.add_column("Argument", style="cyan", no_wrap=True)
        table.add_column("Value", style="magenta")

        for arg in vars(args):
            table.add_row(arg, str(getattr(args, arg)))

        console.print(table)

    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()
            self.show_splash()

            current_directory = os.getcwd()

            config = self._config_manager.load_config(current_directory)
            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.")

                self.print_args_table(args)

                # Load MDAnalysis Universe
                tprfile = args.top_traj_file[0]
                trrfile = args.top_traj_file[1:]
                forcefile = args.force_file
                fileformat = args.file_format
                kcal_units = args.kcal_force_units

                # Create shared UniverseOperations instance
                universe_operations = UniverseOperations()

                if forcefile is None:
                    logger.debug(f"Loading Universe with {tprfile} and {trrfile}")
                    u = mda.Universe(tprfile, trrfile, format=fileformat)
                else:
                    u = universe_operations.merge_forces(
                        tprfile, trrfile, forcefile, fileformat, kcal_units
                    )

                self._config_manager.input_parameters_validation(u, args)

                # Create LevelManager instance
                level_manager = LevelManager(universe_operations)

                # Create GroupMolecules instance
                group_molecules = GroupMolecules()

                # Create shared DihedralAnalysis with injected universe_operations
                dihedral_analysis = DihedralAnalysis(
                    universe_operations=universe_operations
                )

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

                entropy_manager.execute()

            self._logging_config.save_console_log()

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

    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	1✔
2	import os	1✔
3	import pickle	1✔
4
5	import MDAnalysis as mda	1✔
6	import requests	1✔
7	import yaml	1✔
8	from art import text2art	1✔
9	from rich.align import Align	1✔
10	from rich.console import Group	1✔
11	from rich.padding import Padding	1✔
12	from rich.panel import Panel	1✔
13	from rich.rule import Rule	1✔
14	from rich.table import Table	1✔
15	from rich.text import Text	1✔
16
17	from CodeEntropy.config.arg_config_manager import ConfigManager	1✔
18	from CodeEntropy.config.data_logger import DataLogger	1✔
19	from CodeEntropy.config.logging_config import LoggingConfig	1✔
20	from CodeEntropy.dihedral_tools import DihedralAnalysis	1✔
21	from CodeEntropy.entropy import EntropyManager	1✔
22	from CodeEntropy.group_molecules import GroupMolecules	1✔
23	from CodeEntropy.levels import LevelManager	1✔
24	from CodeEntropy.mda_universe_operations import UniverseOperations	1✔
25
26	logger = logging.getLogger(__name__)	1✔
27	console = LoggingConfig.get_console()	1✔
28
29
30	class RunManager:	1✔
31	"""
32	Handles the setup and execution of entropy analysis runs, including configuration
33	loading, logging, and access to physical constants used in calculations.
34	"""
35
36	def __init__(self, folder):	1✔
37	"""
38	Initializes the RunManager with the working folder and sets up configuration,
39	data logging, and logging systems. Also defines physical constants used in
40	entropy calculations.
41	"""
42	self.folder = folder	1✔
43	self._config_manager = ConfigManager()	1✔
44	self._data_logger = DataLogger()	1✔
45	self._logging_config = LoggingConfig(folder)	1✔
46	self._N_AVOGADRO = 6.0221415e23	1✔
47	self._DEF_TEMPER = 298	1✔
48
49	@property	1✔
50	def N_AVOGADRO(self):	1✔
51	"""Returns Avogadro's number used in entropy calculations."""
52	return self._N_AVOGADRO	1✔
53
54	@property	1✔
55	def DEF_TEMPER(self):	1✔
56	"""Returns the default temperature (in Kelvin) used in the analysis."""
57	return self._DEF_TEMPER	1✔
58
59	@staticmethod	1✔
60	def create_job_folder():	1✔
61	"""
62	Create a new job folder with an incremented job number based on existing
63	folders.
64	"""
65	# Get the current working directory
66	current_dir = os.getcwd()	1✔
67
68	# Get a list of existing folders that start with "job"
69	existing_folders = [f for f in os.listdir(current_dir) if f.startswith("job")]	1✔
70
71	# Extract numbers from existing folder names
72	job_numbers = []	1✔
73	for folder in existing_folders:	1✔
74	try:	1✔
75	# Assuming folder names are in the format "jobXXX"
76	job_number = int(folder[3:]) # Get the number part after "job"	1✔
77	job_numbers.append(job_number)	1✔
78	except ValueError:	1✔
79	continue # Ignore any folder names that don't follow the pattern	1✔
80
81	# If no folders exist, start with job001
82	if not job_numbers:	1✔
83	next_job_number = 1	1✔
84	else:
85	next_job_number = max(job_numbers) + 1	1✔
86
87	# Create the new job folder name
88	new_job_folder = f"job{next_job_number:03d}"	1✔
89
90	# Create the full path to the new folder
91	new_folder_path = os.path.join(current_dir, new_job_folder)	1✔
92
93	# Create the directory
94	os.makedirs(new_folder_path, exist_ok=True)	1✔
95
96	# Return the path of the newly created folder
97	return new_folder_path	1✔
98
99	def load_citation_data(self):	1✔
100	"""
101	Load CITATION.cff from GitHub into memory.
102	Return empty dict if offline.
103	"""
104	url = (	1✔
105	"https://raw.githubusercontent.com/CCPBioSim/"
106	"CodeEntropy/refs/heads/main/CITATION.cff"
107	)
108	try:	1✔
109	response = requests.get(url, timeout=10)	1✔
110	response.raise_for_status()	1✔
111	return yaml.safe_load(response.text)	1✔
112	except requests.exceptions.RequestException:	1✔
113	return None	1✔
114
115	def show_splash(self):	1✔
116	"""Render splash screen with optional citation metadata."""
117	citation = self.load_citation_data()	1✔
118
119	if citation:	1✔
120	# ASCII Title
121	ascii_title = text2art(citation.get("title", "CodeEntropy"))	1✔
122	ascii_render = Align.center(Text(ascii_title, style="bold white"))	1✔
123
124	# Metadata
125	version = citation.get("version", "?")	1✔
126	release_date = citation.get("date-released", "?")	1✔
127	url = citation.get("url", citation.get("repository-code", ""))	1✔
128
129	version_text = Align.center(	1✔
130	Text(f"Version {version} \| Released {release_date}", style="green")
131	)
132	url_text = Align.center(Text(url, style="blue underline"))	1✔
133
134	# Description block
135	abstract = citation.get("abstract", "No description available.")	1✔
136	description_title = Align.center(	1✔
137	Text("Description", style="bold magenta underline")
138	)
139	description_body = Align.center(	1✔
140	Padding(Text(abstract, style="white", justify="left"), (0, 4))
141	)
142
143	# Contributors table
144	contributors_title = Align.center(	1✔
145	Text("Contributors", style="bold magenta underline")
146	)
147
148	author_table = Table(	1✔
149	show_header=True, header_style="bold yellow", box=None, pad_edge=False
150	)
151	author_table.add_column("Name", style="bold", justify="center")	1✔
152	author_table.add_column("Affiliation", justify="center")	1✔
153
154	for author in citation.get("authors", []):	1✔
155	name = (	1✔
156	f"{author.get('given-names', '')} {author.get('family-names', '')}"
157	).strip()
158	affiliation = author.get("affiliation", "")	1✔
159	author_table.add_row(name, affiliation)	1✔
160
161	contributors_table = Align.center(Padding(author_table, (0, 4)))	1✔
162
163	# Full layout
164	splash_content = Group(	1✔
165	ascii_render,
166	Rule(style="cyan"),
167	version_text,
168	url_text,
169	Text(),
170	description_title,
171	description_body,
172	Text(),
173	contributors_title,
174	contributors_table,
175	)
176	else:
177	# ASCII Title
178	ascii_title = text2art("CodeEntropy")	1✔
179	ascii_render = Align.center(Text(ascii_title, style="bold white"))	1✔
180
181	splash_content = Group(	1✔
182	ascii_render,
183	)
184
185	splash_panel = Panel(	1✔
186	splash_content,
187	title="[bold bright_cyan]Welcome to CodeEntropy",
188	title_align="center",
189	border_style="bright_cyan",
190	padding=(1, 4),
191	expand=True,
192	)
193
194	console.print(splash_panel)	1✔
195
196	def print_args_table(self, args):	1✔
197	table = Table(title="Run Configuration", expand=True)	1✔
198
199	table.add_column("Argument", style="cyan", no_wrap=True)	1✔
200	table.add_column("Value", style="magenta")	1✔
201
202	for arg in vars(args):	1✔
203	table.add_row(arg, str(getattr(args, arg)))	1✔
204
205	console.print(table)	1✔
206
207	def run_entropy_workflow(self):	1✔
208	"""
209	Runs the entropy analysis workflow by setting up logging, loading configuration
210	files, parsing arguments, and executing the analysis for each configured run.
211	Initializes the MDAnalysis Universe and supporting managers, and logs all
212	relevant inputs and commands.
213	"""
214	try:	1✔
215	logger = self._logging_config.setup_logging()	1✔
216	self.show_splash()	1✔
217
218	current_directory = os.getcwd()	1✔
219
220	config = self._config_manager.load_config(current_directory)	1✔
221	parser = self._config_manager.setup_argparse()	1✔
222	args, _ = parser.parse_known_args()	1✔
223	args.output_file = os.path.join(self.folder, args.output_file)	1✔
224
225	for run_name, run_config in config.items():	1✔
226	if not isinstance(run_config, dict):	1✔
227	logger.warning(	1✔
228	f"Run configuration for {run_name} is not a dictionary."
229	)
230	continue	1✔
231
232	args = self._config_manager.merge_configs(args, run_config)	1✔
233
234	log_level = logging.DEBUG if args.verbose else logging.INFO	1✔
235	self._logging_config.update_logging_level(log_level)	1✔
236
237	command = " ".join(os.sys.argv)	1✔
238	logging.getLogger("commands").info(command)	1✔
239
240	if not getattr(args, "top_traj_file", None):	1✔
241	raise ValueError("Missing 'top_traj_file' argument.")	1✔
242	if not getattr(args, "selection_string", None):	1✔
243	raise ValueError("Missing 'selection_string' argument.")	1✔
244
245	self.print_args_table(args)	1✔
246
247	# Load MDAnalysis Universe
248	tprfile = args.top_traj_file[0]	1✔
249	trrfile = args.top_traj_file[1:]	1✔
250	forcefile = args.force_file	1✔
251	fileformat = args.file_format	1✔
252	kcal_units = args.kcal_force_units	1✔
253
254	# Create shared UniverseOperations instance
255	universe_operations = UniverseOperations()	1✔
256
257	if forcefile is None:	1✔
258	logger.debug(f"Loading Universe with {tprfile} and {trrfile}")	1✔
259	u = mda.Universe(tprfile, trrfile, format=fileformat)	1✔
260	else:
NEW 261	u = universe_operations.merge_forces(	×
262	tprfile, trrfile, forcefile, fileformat, kcal_units
263	)
264
265	self._config_manager.input_parameters_validation(u, args)	1✔
266
267	# Create LevelManager instance
268	level_manager = LevelManager(universe_operations)	1✔
269
270	# Create GroupMolecules instance
271	group_molecules = GroupMolecules()	1✔
272
273	# Create shared DihedralAnalysis with injected universe_operations
274	dihedral_analysis = DihedralAnalysis(	1✔
275	universe_operations=universe_operations
276	)
277
278	# Inject all dependencies into EntropyManager
279	entropy_manager = EntropyManager(	1✔
280	run_manager=self,
281	args=args,
282	universe=u,
283	data_logger=self._data_logger,
284	level_manager=level_manager,
285	group_molecules=group_molecules,
286	dihedral_analysis=dihedral_analysis,
287	universe_operations=universe_operations,
288	)
289
290	entropy_manager.execute()	1✔
291
292	self._logging_config.save_console_log()	1✔
293
294	except Exception as e:	1✔
295	logger.error(f"RunManager encountered an error: {e}", exc_info=True)	1✔
296	raise	1✔
297
298	def write_universe(self, u, name="default"):	1✔
299	"""Write a universe to working directories as pickle
300
301	Parameters
302	----------
303	u : MDAnalyse.Universe
304	A Universe object will all topology, dihedrals,coordinates and force
305	information
306	name : str, Optional. default: 'default'
307	The name of file with sub file name .pkl
308
309	Returns
310	-------
311	name : str
312	filename of saved universe
313	"""
314	filename = f"{name}.pkl"	1✔
315	pickle.dump(u, open(filename, "wb"))	1✔
316	return name	1✔
317
318	def read_universe(self, path):	1✔
319	"""read a universe to working directories as pickle
320
321	Parameters
322	----------
323	path : str
324	The path to file.
325
326	Returns
327	-------
328	u : MDAnalysis.Universe
329	A Universe object will all topology, dihedrals,coordinates and force
330	information.
331	"""
332	u = pickle.load(open(path, "rb"))	1✔
333	return u	1✔
334
335	def change_lambda_units(self, arg_lambdas):	1✔
336	"""Unit of lambdas : kJ2 mol-2 A-2 amu-1
337	change units of lambda to J/s2"""
338	# return arg_lambdas * N_AVOGADRO * N_AVOGADRO * AMU2KG * 1e-26
339	return arg_lambdas * 1e29 / self.N_AVOGADRO	1✔
340
341	def get_KT2J(self, arg_temper):	1✔
342	"""A temperature dependent KT to Joule conversion"""
343	return 4.11e-21 * arg_temper / self.DEF_TEMPER	1✔

CCPBioSim / CodeEntropy / 19675730718

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