"""Handles the analyze subcommand.
Author: Cameron F. Abrams <cfa22@drexel.edu>
"""
import json
import logging
import os
from pathlib import Path
import yaml
from ..core import projectfilesystem as pfs
from ..core.configuration import Configuration
from ..external import software as software
from ..external.gromacs import gmx_command
from ..utils.logsetup import setup_logging
logger=logging.getLogger(__name__)
[docs]
class Analyze:
allowed_keys=['gromacs','command','subdir','options','links','outfile','console-input','matchlines']
required_keys=['command','subdir']
default_params={
'gromacs' : {
'gmx': 'gmx'
},
}
def __init__(self,indict,strict=True):
self.params={}
for p,v in self.default_params.items():
self.params[p]=indict.get(p,v)
for p,v in indict.items():
if not p in self.allowed_keys:
if strict:
logger.info(f'Ignoring directive \'{p}\' in yaml input file')
if p in self.default_params:
logger.info(f'Overwriting default {p} value')
self.params[p]=v
self.console_output=None
[docs]
def do(self,**gromacs_dict):
"""Handles executing the analysis."""
p=self.params
print(p)
for rk in self.required_keys:
assert rk in p, f'Error: no {rk} value found'
# logger.info(f'do {p}')
# if a gromacs dict is passed in, assume this overrides the one read in from the file
if gromacs_dict:
software.set_gmx_preferences(gromacs_dict)
else:
software.set_gmx_preferences(p['gromacs'])
logger.info(f'going to {p["subdir"]}')
pfs.go_to(p['subdir'])
# make symlinks to requested files
symlinks=p.get('links',[])
for input_file in symlinks:
srcnm=os.path.join(pfs.proj(),input_file)
bsnm=os.path.basename(srcnm)
chk=Path(bsnm)
if not chk.is_symlink():
os.symlink(srcnm,bsnm)
else:
logger.info(f'Symlink {bsnm} already exists.')
cfile=''
if 'console-input' in p:
cfile='console-in.txt'
ci=p['console-input']
with open(cfile,'w') as f:
for ch in ci:
f.write(ch+'\n')
self.console_output=gmx_command(p['command'],p.get('options',{}),console_in=cfile)
logger.info(f'Command {p["command"]} completed.')
[docs]
def parse_console_output(self):
if not self.console_output:
logger.info(f'No console output')
return
p=self.params
# either we are grepping out lines from console output or putting it all out there
if not 'outfile' in p:
logger.info(f'Here is the console output')
logger.info(self.console_output)
else:
if not 'matchlines' in p:
with open(p['outfile'],'w') as f:
f.write(self.console_output)
else:
svlns=[]
console_lines=self.console_output.split('\n')
for cl in console_lines:
for ml in p['matchlines']:
if ml in cl:
svlns.append(cl)
with open(p['outfile'],'w') as f:
for s in svlns:
f.write(s+'\n')
logger.info(f'Created {p["outfile"]} in {p["subdir"]}')
[docs]
class AnalyzeDensity(Analyze):
""" Analyze class for handling trajectory density profile calculation
"""
default_params={
'subdir': 'analyze/density',
'links': [f'{pfs.Dirs.postsim}/equilibrate/equilibrate.tpr',f'{pfs.Dirs.postsim}/equilibrate/equilibrate.trr'],
'gromacs' : {
'gmx': 'gmx'
},
'command': 'density',
'options': {
's':'equilibrate.tpr',
'f':'equilibrate.trr',
'o':'density.xvg',
'xvg': 'none',
'b': 0,
'd': 'Z',
'sl': 50
},
'console-input': ['0']
}
[docs]
class AnalyzeFFV(Analyze):
default_params={
'subdir': 'analyze/freevolume',
'links': [f'{pfs.Dirs.postsim}/equilibrate/equilibrate.tpr',f'{pfs.Dirs.postsim}/equilibrate/equilibrate.trr'],
'gromacs' : {
'gmx': 'gmx'
},
'command': 'freevolume',
'options': {
's':'equilibrate.tpr',
'f':'equilibrate.trr',
'o':'ffv.xvg',
'xvg': 'none',
'b': 0.0
},
'outfile': 'ffv.dat',
'matchlines': ['Free volume','Total volume','Number of molecules','Average molar mass','Density','Molecular volume Vm assuming homogeneity:','Molecular van der Waals volume assuming homogeneity:','Fractional free volume']
}
[docs]
class AnalyzeConfiguration:
""" handles reading and parsing an analysis input config file.
Config file format
- { key1: {<paramdict>}}
- { key2: {<paramdict>}}
...
The config file is a list of single-element dictionaries, whose single keyword
indicates the type of analysis to be run; analyses are run in the order
they appear in the config file.
"""
default_class=Analyze
predefined_classes={'density':AnalyzeDensity,'freevolume':AnalyzeFFV}
def __init__(self):
self.cfgFile=''
self.baselist=[]
self.stagelist=[]
[docs]
@classmethod
def read(cls,filename,parse=True,**kwargs):
"""Generates a new PostsimConfiguration object by reading in the JSON or YAML file indicated by filename.
Args:
filename (str): name of file from which to read new PostsimConfiguration object
parse (bool): if True, parse the input configuration file, defaults to True
Raises:
Exception: if extension of filename is not '.json' or '.yaml' or '.yml'
Returns:
PostsimConfiguration: a new PostsimConfiguration object
"""
basename,extension=os.path.splitext(filename)
if extension=='.json':
return cls._read_json(filename,parse,**kwargs)
elif extension=='.yaml' or extension=='.yml':
return cls._read_yaml(filename,parse,**kwargs)
else:
raise Exception(f'Unknown config file extension {extension}')
@classmethod
def _read_json(cls,filename,parse=True,**kwargs):
"""Creates a new PostsimConfiguration object by reading from JSON input.
Args:
filename (str): name of JSON file
parse (bool): if True, parse the JSON data, defaults to True
Returns:
PostsimConfiguration: a new PostsimConfiguration object
"""
inst=cls()
inst.cfgFile=filename
with open(filename,'r') as f:
inst.baselist=json.load(f)
assert type(inst.baselist)==list,f'Poorly formatted {filename}'
if parse: inst.parse(**kwargs)
return inst
@classmethod
def _read_yaml(cls,filename,parse=True,**kwargs):
"""Creates a new PostsimConfiguration object by reading from YAML input.
Args:
filename (str): name of YAML file
parse (bool): if True, parse the YAML data, defaults to True
Returns:
PostsimConfiguration: a new PostsimConfiguration object
"""
inst=cls()
inst.cfgFile=filename
with open(filename,'r') as f:
inst.baselist=yaml.safe_load(f)
assert type(inst.baselist)==list,f'Poorly formatted {filename}'
if parse: inst.parse(**kwargs)
return inst
[docs]
def parse(self,**kwargs):
"""Parses a PostsimConfiguration file to build the list of stages to run."""
for content in self.baselist:
analysistype=content['command']
if analysistype in self.predefined_classes:
self.stagelist.append(self.predefined_classes[analysistype](content))
else:
self.stagelist.append(self.default_class(content))
[docs]
def analyze(args):
"""Handles the analyze subcommand for managing gromacs-based trajectory analyses.
Args:
args (argparse.Namespace): command-line arguments
"""
setup_logging(args.loglevel, no_banner=args.no_banner)
ess='y' if len(args.proj)==0 else 'ies'
ogromacs={}
if args.ocfg:
ocfg=Configuration.read(args.ocfg)
ogromacs=ocfg.gromacs
cfg=AnalyzeConfiguration.read(args.cfg)
logger.debug(f'{cfg.baselist}')
logger.info(f'Project director{ess}: {args.proj}')
software.sw_setup()
logger.debug(f'ogromacs {ogromacs}')
for d in args.proj:
pfs.pfs_setup(root=os.getcwd(),topdirs=pfs.Dirs.analyze_topdirs,verbose=True,projdir=d,reProject=False,userlibrary=args.lib)
pfs.go_to(pfs.Dirs.analyze)
for stage in cfg.stagelist:
stage.do(**ogromacs)
stage.parse_console_output()
pfs.go_root()