19166661659

Committed 07 Nov 2025 11:13AM UTC coverage: 98.881% (-1.1%) from 100.0%

Build # 19166661659

Build Type

Pull #175

github

Committed by

web-flow

Commit Message

Merge 1ffc3c313 into 71a7253d9

Pull Request Pull Request #175: 172 coordinates and forces in separate files

Coverage Stats

7 of 19 new or added lines in 2 files covered. (36.84%)

1060 of 1072 relevant lines covered (98.88%)

8.9 hits per line

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

93.37

/CodeEntropy/run.py

import logging
import os
import pickle

import MDAnalysis as mda
import requests
import yaml
from art import text2art
from MDAnalysis.analysis.base import AnalysisFromFunction
from MDAnalysis.coordinates.memory import MemoryReader
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.entropy import EntropyManager
from CodeEntropy.group_molecules import GroupMolecules
from CodeEntropy.levels import LevelManager

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:]
                fileformat = args.file_format
                logger.debug(f"Loading Universe with {tprfile} and {trrfile}")
                u = mda.Universe(tprfile, trrfile, format=fileformat)

                # If forces are in separate file merge them with the
                # coordinates from the trajectory file
                forcefile = args.force_file
                if forcefile is not None:
                    logger.debug(f"Loading Universe with {forcefile}")
                    u_force = mda.Universe(tprfile, forcefile, format=fileformat)
                    select_atom = u.select_atoms("all")
                    select_atom_force = u_force.select_atoms("all")

                    coordinates = (
                        AnalysisFromFunction(
                            lambda ag: ag.positions.copy(), select_atom
                        )
                        .run()
                        .results["timeseries"]
                    )
                    forces = (
                        AnalysisFromFunction(
                            lambda ag: ag.positions.copy(), select_atom_force
                        )
                        .run()
                        .results["timeseries"]
                    )

                    if args.kcal_force_units:
                        # Convert from kcal to kJ
                        forces *= 4.184

                    logger.debug("Merging forces with coordinates universe.")
                    new_universe = mda.Merge(select_atom)
                    new_universe.load_new(coordinates, forces=forces)

                    u = new_universe

                self._config_manager.input_parameters_validation(u, args)

                # Create LevelManager instance
                level_manager = LevelManager()

                # Create GroupMolecules instance
                group_molecules = GroupMolecules()

                # 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,
                )

                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 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]
        )
        u2 = mda.Merge(select_atom)
        u2.load_new(coordinates, format=MemoryReader, forces=forces)
        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"]
        )
        u2 = mda.Merge(select_atom)
        u2.load_new(coordinates, format=MemoryReader, forces=forces)
        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	9✔
2	import os	9✔
3	import pickle	9✔
4
5	import MDAnalysis as mda	9✔
6	import requests	9✔
7	import yaml	9✔
8	from art import text2art	9✔
9	from MDAnalysis.analysis.base import AnalysisFromFunction	9✔
10	from MDAnalysis.coordinates.memory import MemoryReader	9✔
11	from rich.align import Align	9✔
12	from rich.console import Group	9✔
13	from rich.padding import Padding	9✔
14	from rich.panel import Panel	9✔
15	from rich.rule import Rule	9✔
16	from rich.table import Table	9✔
17	from rich.text import Text	9✔
18
19	from CodeEntropy.config.arg_config_manager import ConfigManager	9✔
20	from CodeEntropy.config.data_logger import DataLogger	9✔
21	from CodeEntropy.config.logging_config import LoggingConfig	9✔
22	from CodeEntropy.entropy import EntropyManager	9✔
23	from CodeEntropy.group_molecules import GroupMolecules	9✔
24	from CodeEntropy.levels import LevelManager	9✔
25
26	logger = logging.getLogger(__name__)	9✔
27	console = LoggingConfig.get_console()	9✔
28
29
30	class RunManager:	9✔
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):	9✔
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	9✔
43	self._config_manager = ConfigManager()	9✔
44	self._data_logger = DataLogger()	9✔
45	self._logging_config = LoggingConfig(folder)	9✔
46	self._N_AVOGADRO = 6.0221415e23	9✔
47	self._DEF_TEMPER = 298	9✔
48
49	@property	9✔
50	def N_AVOGADRO(self):	9✔
51	"""Returns Avogadro's number used in entropy calculations."""
52	return self._N_AVOGADRO	9✔
53
54	@property	9✔
55	def DEF_TEMPER(self):	9✔
56	"""Returns the default temperature (in Kelvin) used in the analysis."""
57	return self._DEF_TEMPER	9✔
58
59	@staticmethod	9✔
60	def create_job_folder():	9✔
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()	9✔
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")]	9✔
70
71	# Extract numbers from existing folder names
72	job_numbers = []	9✔
73	for folder in existing_folders:	9✔
74	try:	9✔
75	# Assuming folder names are in the format "jobXXX"
76	job_number = int(folder[3:]) # Get the number part after "job"	9✔
77	job_numbers.append(job_number)	9✔
78	except ValueError:	9✔
79	continue # Ignore any folder names that don't follow the pattern	9✔
80
81	# If no folders exist, start with job001
82	if not job_numbers:	9✔
83	next_job_number = 1	9✔
84	else:
85	next_job_number = max(job_numbers) + 1	9✔
86
87	# Create the new job folder name
88	new_job_folder = f"job{next_job_number:03d}"	9✔
89
90	# Create the full path to the new folder
91	new_folder_path = os.path.join(current_dir, new_job_folder)	9✔
92
93	# Create the directory
94	os.makedirs(new_folder_path, exist_ok=True)	9✔
95
96	# Return the path of the newly created folder
97	return new_folder_path	9✔
98
99	def load_citation_data(self):	9✔
100	"""
101	Load CITATION.cff from GitHub into memory.
102	Return empty dict if offline.
103	"""
104	url = (	9✔
105	"https://raw.githubusercontent.com/CCPBioSim/"
106	"CodeEntropy/refs/heads/main/CITATION.cff"
107	)
108	try:	9✔
109	response = requests.get(url, timeout=10)	9✔
110	response.raise_for_status()	9✔
111	return yaml.safe_load(response.text)	9✔
112	except requests.exceptions.RequestException:	9✔
113	return None	9✔
114
115	def show_splash(self):	9✔
116	"""Render splash screen with optional citation metadata."""
117	citation = self.load_citation_data()	9✔
118
119	if citation:	9✔
120	# ASCII Title
121	ascii_title = text2art(citation.get("title", "CodeEntropy"))	9✔
122	ascii_render = Align.center(Text(ascii_title, style="bold white"))	9✔
123
124	# Metadata
125	version = citation.get("version", "?")	9✔
126	release_date = citation.get("date-released", "?")	9✔
127	url = citation.get("url", citation.get("repository-code", ""))	9✔
128
129	version_text = Align.center(	9✔
130	Text(f"Version {version} \| Released {release_date}", style="green")
131	)
132	url_text = Align.center(Text(url, style="blue underline"))	9✔
133
134	# Description block
135	abstract = citation.get("abstract", "No description available.")	9✔
136	description_title = Align.center(	9✔
137	Text("Description", style="bold magenta underline")
138	)
139	description_body = Align.center(	9✔
140	Padding(Text(abstract, style="white", justify="left"), (0, 4))
141	)
142
143	# Contributors table
144	contributors_title = Align.center(	9✔
145	Text("Contributors", style="bold magenta underline")
146	)
147
148	author_table = Table(	9✔
149	show_header=True, header_style="bold yellow", box=None, pad_edge=False
150	)
151	author_table.add_column("Name", style="bold", justify="center")	9✔
152	author_table.add_column("Affiliation", justify="center")	9✔
153
154	for author in citation.get("authors", []):	9✔
155	name = (	9✔
156	f"{author.get('given-names', '')} {author.get('family-names', '')}"
157	).strip()
158	affiliation = author.get("affiliation", "")	9✔
159	author_table.add_row(name, affiliation)	9✔
160
161	contributors_table = Align.center(Padding(author_table, (0, 4)))	9✔
162
163	# Full layout
164	splash_content = Group(	9✔
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")	9✔
179	ascii_render = Align.center(Text(ascii_title, style="bold white"))	9✔
180
181	splash_content = Group(	9✔
182	ascii_render,
183	)
184
185	splash_panel = Panel(	9✔
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)	9✔
195
196	def print_args_table(self, args):	9✔
197	table = Table(title="Run Configuration", expand=True)	9✔
198
199	table.add_column("Argument", style="cyan", no_wrap=True)	9✔
200	table.add_column("Value", style="magenta")	9✔
201
202	for arg in vars(args):	9✔
203	table.add_row(arg, str(getattr(args, arg)))	9✔
204
205	console.print(table)	9✔
206
207	def run_entropy_workflow(self):	9✔
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:	9✔
215	logger = self._logging_config.setup_logging()	9✔
216	self.show_splash()	9✔
217
218	current_directory = os.getcwd()	9✔
219
220	config = self._config_manager.load_config(current_directory)	9✔
221	parser = self._config_manager.setup_argparse()	9✔
222	args, _ = parser.parse_known_args()	9✔
223	args.output_file = os.path.join(self.folder, args.output_file)	9✔
224
225	for run_name, run_config in config.items():	9✔
226	if not isinstance(run_config, dict):	9✔
227	logger.warning(	9✔
228	f"Run configuration for {run_name} is not a dictionary."
229	)
230	continue	9✔
231
232	args = self._config_manager.merge_configs(args, run_config)	9✔
233
234	log_level = logging.DEBUG if args.verbose else logging.INFO	9✔
235	self._logging_config.update_logging_level(log_level)	9✔
236
237	command = " ".join(os.sys.argv)	9✔
238	logging.getLogger("commands").info(command)	9✔
239
240	if not getattr(args, "top_traj_file", None):	9✔
241	raise ValueError("Missing 'top_traj_file' argument.")	9✔
242	if not getattr(args, "selection_string", None):	9✔
243	raise ValueError("Missing 'selection_string' argument.")	9✔
244
245	self.print_args_table(args)	9✔
246
247	# Load MDAnalysis Universe
248	tprfile = args.top_traj_file[0]	9✔
249	trrfile = args.top_traj_file[1:]	9✔
250	fileformat = args.file_format	9✔
251	logger.debug(f"Loading Universe with {tprfile} and {trrfile}")	9✔
252	u = mda.Universe(tprfile, trrfile, format=fileformat)	9✔
253
254	# If forces are in separate file merge them with the
255	# coordinates from the trajectory file
256	forcefile = args.force_file	9✔
257	if forcefile is not None:	9✔
NEW 258	logger.debug(f"Loading Universe with {forcefile}")	×
NEW 259	u_force = mda.Universe(tprfile, forcefile, format=fileformat)	×
NEW 260	select_atom = u.select_atoms("all")	×
NEW 261	select_atom_force = u_force.select_atoms("all")	×
262
NEW 263	coordinates = (	×
264	AnalysisFromFunction(
265	lambda ag: ag.positions.copy(), select_atom
266	)
267	.run()
268	.results["timeseries"]
269	)
NEW 270	forces = (	×
271	AnalysisFromFunction(
272	lambda ag: ag.positions.copy(), select_atom_force
273	)
274	.run()
275	.results["timeseries"]
276	)
277
NEW 278	if args.kcal_force_units:	×
279	# Convert from kcal to kJ
NEW 280	forces *= 4.184	×
281
NEW 282	logger.debug("Merging forces with coordinates universe.")	×
NEW 283	new_universe = mda.Merge(select_atom)	×
NEW 284	new_universe.load_new(coordinates, forces=forces)	×
285
NEW 286	u = new_universe	×
287
288	self._config_manager.input_parameters_validation(u, args)	9✔
289
290	# Create LevelManager instance
291	level_manager = LevelManager()	9✔
292
293	# Create GroupMolecules instance
294	group_molecules = GroupMolecules()	9✔
295
296	# Inject all dependencies into EntropyManager
297	entropy_manager = EntropyManager(	9✔
298	run_manager=self,
299	args=args,
300	universe=u,
301	data_logger=self._data_logger,
302	level_manager=level_manager,
303	group_molecules=group_molecules,
304	)
305
306	entropy_manager.execute()	9✔
307
308	self._logging_config.save_console_log()	9✔
309
310	except Exception as e:	9✔
311	logger.error(f"RunManager encountered an error: {e}", exc_info=True)	9✔
312	raise	9✔
313
314	def new_U_select_frame(self, u, start=None, end=None, step=1):	9✔
315	"""Create a reduced universe by dropping frames according to user selection
316
317	Parameters
318	----------
319	u : MDAnalyse.Universe
320	A Universe object will all topology, dihedrals,coordinates and force
321	information
322	start : int or None, Optional, default: None
323	Frame id to start analysis. Default None will start from frame 0
324	end : int or None, Optional, default: None
325	Frame id to end analysis. Default None will end at last frame
326	step : int, Optional, default: 1
327	Steps between frame.
328
329	Returns
330	-------
331	u2 : MDAnalysis.Universe
332	reduced universe
333	"""
334	if start is None:	9✔
335	start = 0	9✔
336	if end is None:	9✔
337	end = len(u.trajectory)	9✔
338	select_atom = u.select_atoms("all", updating=True)	9✔
339	coordinates = (	9✔
340	AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
341	.run()
342	.results["timeseries"][start:end:step]
343	)
344	forces = (	9✔
345	AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
346	.run()
347	.results["timeseries"][start:end:step]
348	)
349	u2 = mda.Merge(select_atom)	9✔
350	u2.load_new(coordinates, format=MemoryReader, forces=forces)	9✔
351	logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")	9✔
352	return u2	9✔
353
354	def new_U_select_atom(self, u, select_string="all"):	9✔
355	"""Create a reduced universe by dropping atoms according to user selection
356
357	Parameters
358	----------
359	u : MDAnalyse.Universe
360	A Universe object will all topology, dihedrals,coordinates and force
361	information
362	select_string : str, Optional, default: 'all'
363	MDAnalysis.select_atoms selection string.
364
365	Returns
366	-------
367	u2 : MDAnalysis.Universe
368	reduced universe
369
370	"""
371	select_atom = u.select_atoms(select_string, updating=True)	9✔
372	coordinates = (	9✔
373	AnalysisFromFunction(lambda ag: ag.positions.copy(), select_atom)
374	.run()
375	.results["timeseries"]
376	)
377	forces = (	9✔
378	AnalysisFromFunction(lambda ag: ag.forces.copy(), select_atom)
379	.run()
380	.results["timeseries"]
381	)
382	u2 = mda.Merge(select_atom)	9✔
383	u2.load_new(coordinates, format=MemoryReader, forces=forces)	9✔
384	logger.debug(f"MDAnalysis.Universe - reduced universe: {u2}")	9✔
385	return u2	9✔
386
387	def write_universe(self, u, name="default"):	9✔
388	"""Write a universe to working directories as pickle
389
390	Parameters
391	----------
392	u : MDAnalyse.Universe
393	A Universe object will all topology, dihedrals,coordinates and force
394	information
395	name : str, Optional. default: 'default'
396	The name of file with sub file name .pkl
397
398	Returns
399	-------
400	name : str
401	filename of saved universe
402	"""
403	filename = f"{name}.pkl"	9✔
404	pickle.dump(u, open(filename, "wb"))	9✔
405	return name	9✔
406
407	def read_universe(self, path):	9✔
408	"""read a universe to working directories as pickle
409
410	Parameters
411	----------
412	path : str
413	The path to file.
414
415	Returns
416	-------
417	u : MDAnalysis.Universe
418	A Universe object will all topology, dihedrals,coordinates and force
419	information.
420	"""
421	u = pickle.load(open(path, "rb"))	9✔
422	return u	9✔
423
424	def change_lambda_units(self, arg_lambdas):	9✔
425	"""Unit of lambdas : kJ2 mol-2 A-2 amu-1
426	change units of lambda to J/s2"""
427	# return arg_lambdas * N_AVOGADRO * N_AVOGADRO * AMU2KG * 1e-26
428	return arg_lambdas * 1e29 / self.N_AVOGADRO	9✔
429
430	def get_KT2J(self, arg_temper):	9✔
431	"""A temperature dependent KT to Joule conversion"""
432	return 4.11e-21 * arg_temper / self.DEF_TEMPER	9✔

CCPBioSim / CodeEntropy / 19166661659

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