##################################################################
## (c) Copyright 2015- by Jaron T. Krogel ##
##################################################################
#====================================================================#
# simulation.py #
# Provides base classes for simulation objects, including input #
# and analysis. The Simulation base class enables a large amount #
# of the functionality of Nexus, including workflow construction #
# and monitoring, in tandem with the ProjectManager class. #
# #
# Content summary: #
# SimulationInput #
# Abstract base class for simulation input. #
# #
# SimulationAnalyzer #
# Abstract base class for simulation data analysis. #
# #
# Simulation #
# Major Nexus class representing a simulation prior to, during, #
# and after execution. Checks dependencies between simulations #
# connected in workflows, manages input file writing, #
# participates in job submission, checks for successful #
# simulation completion, and analyzes output data. Saves state #
# image of simulation progress regularly. Contains #
# SimulationInput, SimulationAnalyzer, and Job objects (also #
# optionally contains a PhysicalSystem object). Derived #
# classes tailor specific functions such as passing dependency #
# data and checking simulation state to a target simulation #
# code. Derived classes include Qmcpack, Pwscf, Vasp, Gamess, #
# Convert4qmc, Pw2qmcpack, SimulationBundle, and #
# TemplateSimulation. #
# #
# NullSimulationInput #
# Simulation input class intended for codes that do not use an #
# input file. #
# #
# NullSimulationAnalyzer #
# Simulation input class intended for codes that do not produce #
# or need to analyze output data. #
# #
# SimulationInputTemplate #
# Supports template input files. A template input file is a #
# standard text input file provided by the user that optionally #
# has specially marked keywords. Using find and replace #
# operations, Nexus can produce variations on the template #
# input file (e.g. to scan over a parameter). In this way #
# Nexus can drive codes that do not have specialized classes #
# derived from Simulation, SimulationInput, or #
# SimulationAnalyzer. #
# #
# SimulationInputMultiTemplate #
# Supports templated input files for codes that take many #
# different files as input (VASP is an example of this). The #
# multi-template is essentially a collection of individual #
# template input files. #
# #
# input_template #
# User-facing function to create SimulationInputTemplate's. #
# #
# multi_input_template #
# User-facing function to create SimulationInputMultiTemplate's.#
# #
#====================================================================#
import os
import sys
import shutil
from pathlib import Path
from string import Template
from subprocess import Popen
import tempfile
from .developer import obj, unavailable, DevBase
from .structure import Structure, read_structure
from .physical_system import PhysicalSystem
from .machines import Job, Workstation, get_machine
from .pseudopotential import ppset
from .nexus_base import NexusCore, nexus_core, dynamic_storage
from .utilities import path_string
#end def return_system
#end class SimulationInput
[docs]
class SimulationAnalyzer(NexusCore):
def __init__(self,sim):
self.not_implemented()
#end def __init__
[docs]
def analyze(self):
self.not_implemented()
#end def analyze
#end class SimulationAnalyzer
[docs]
class SimulationEmulator(NexusCore):
[docs]
def run(self):
self.not_implemented()
#end def run
#end class SimulationEmulator
[docs]
class SimulationImage(NexusCore):
save_only_fields = set([
# user block (temporary) of (sim+) subcascade
'block',
'block_subcascade',
# local/remote/results directories
'locdir',
'remdir',
'resdir',
# image directories
'imlocdir',
'imremdir',
'imresdir',
])
load_fields = set([
# important sim variables
'identifier',
'path',
'process_id',
# properties of the executable
'app_name',
'app_props',
# names of in/out/err files
'infile',
'outfile',
'errfile',
# directory and image file names for sim/input/analyzer
'image_dir',
'sim_image',
'input_image',
'analyzer_image',
# files copied in/out before/after run
'files',
'outputs',
# simulation status flags
'setup',
'sent_files',
'submitted',
'finished',
'failed',
'got_output',
'analyzed',
# cascade status flag
'subcascade_finished',
])
save_fields = load_fields | save_only_fields
def __init__(self):
None
#end def __init__
[docs]
def save_image(self,sim,imagefile):
self.clear()
self.transfer_from(sim,SimulationImage.save_fields)
self.save(imagefile)
self.clear()
#end def save_image
[docs]
def load_image(self,sim,imagefile):
self.clear()
self.load(imagefile)
self.transfer_to(sim,SimulationImage.load_fields)
self.clear()
#end def load_image
#end class SimulationImage
[docs]
class Simulation(NexusCore):
input_type = SimulationInput
analyzer_type = SimulationAnalyzer
generic_identifier = 'sim'
infile_extension = '.in'
outfile_extension = '.out'
errfile_extension = '.err'
application = 'simapp'
application_properties = set(['serial'])
application_results = set()
allow_overlapping_files = False
allowed_inputs = set(['identifier','path','infile','outfile','errfile','imagefile',
'input','job','files','dependencies','analysis_request',
'block','block_subcascade','app_name','app_props','system',
'skip_submit','force_write','simlabel','fake_sim',
'restartable','force_restart'])
sim_imagefile = 'sim.p'
input_imagefile = 'input.p'
analyzer_imagefile = 'analyzer.p'
image_directory = 'sim'
supports_restarts = False
renew_app_command = False
is_bundle = False
sim_count = 0
creating_fake_sims = False
sim_directories = dict()
all_sims = []
[docs]
@classmethod
def clear_all_sims(cls):
cls.sim_directories.clear()
cls.all_sims = []
cls.sim_count = 0
#end def clear_all_sims
[docs]
@classmethod
def code_name(cls):
return cls.generic_identifier
#end def code_name
# test needed
#end def separate_inputs
def __init__(self,**kwargs):
#user specified variables
self.path = '' #directory where sim will be run
self.job = None #Job object for machine
self.dependencies = obj() #Simulation results on which sim serially depends
self.restartable = False #if True, job can be automatically restarted as deemed appropriate
self.force_restart = False #force a restart of the run
#variables determined by self
self.identifier = self.generic_identifier
self.simid = Simulation.sim_count
self.simlabel = None
Simulation.sim_count+=1
self.files = set()
self.app_name = self.application
self.app_props = list(self.application_properties)
self.sim_image = self.sim_imagefile
self.input_image = self.input_imagefile
self.analyzer_image = self.analyzer_imagefile
self.image_dir = self.image_directory
self.input = self.input_type()
self.system = None
self.dependents = obj()
self.created_directories = False
self.got_dependencies = False
self.setup = False
self.sent_files = False
self.submitted = False
self.finished = False
self.failed = False
self.got_output = False
self.analyzed = False
self.subcascade_finished = False
self.dependency_ids = set()
self.wait_ids = set()
self.block = False
self.block_subcascade = False
self.skip_submit = nexus_core.skip_submit
self.force_write = False
self.loaded = False
self.ordered_dependencies = []
self.process_id = None
self.infile = None
self.outfile = None
self.errfile = None
self.bundleable = True
self.bundled = False
self.bundler = None
self.fake_sim = Simulation.creating_fake_sims
#variables determined by derived classes
self.outputs = None #object representing output data
# accessed by dependents when calling get_dependencies
self.set(**kwargs)
self.pre_init()
self.set_directories()
self.set_files()
self.propagate_identifier()
if len(kwargs)>0:
self.init_job()
#end if
self.post_init()
Simulation.all_sims.append(self)
# dynamic workflow support
if nexus_core.dynamic:
assert self.simid not in dynamic_storage.simulation_ids
self.produces = set()
self.products = obj()
self.filled_products = False
self.fill_produces()
for prod in self.produces:
self.products[prod] = None
dynamic_storage.simulations[self.simid] = self
dynamic_storage.simulation_ids.add(self.simid)
# instantly restore image/state data from disk
self.reconstruct_cascade()
if self.finished:
self.fill_products()
#end def __init__
[docs]
def fake(self):
return self.fake_sim
#end def fake
[docs]
def init_job(self):
if self.job is None:
self.error('job not provided. Input field job must be set to a Job object.')
elif not isinstance(self.job,Job):
self.error('Input field job must be set to a Job object\nyou provided an object of type: {0}\nwith value: {1}'.format(self.job.__class__.__name__,self.job))
#end if
self.job = self.job.copy()
self.init_job_extra()
self.job.initialize(self)
#end def init_job
#end def init_job_extra
[docs]
def set_app_name(self,app_name):
self.app_name = app_name
#end def set_app_name
[docs]
def set(self,**kw):
cls = self.__class__
if 'dependencies' in kw:
self.depends(*kw['dependencies'])
del kw['dependencies']
#end if
kwset = set(kw.keys())
invalid = kwset - self.allowed_inputs
if len(invalid)>0:
self.error('received invalid inputs\ninvalid inputs: {0}\nallowed inputs are: {1}'.format(sorted(invalid),sorted(self.allowed_inputs)))
#end if
allowed = kwset & self.allowed_inputs
for name in allowed:
self[name] = kw[name]
#end for
if 'path' in allowed:
if not isinstance(self.path, str | Path):
self.error('path must be a string or Path, you provided {0} (type {1})'.format(self.path,self.path.__class__.__name__))
else:
self.path = path_string(self.path)
p = self.path
#end if
if p.startswith('./'):
p = p[2:]
#end if
ld = nexus_core.local_directory
if p.startswith(ld):
p = p.split(ld)[1].lstrip('/')
#end if
self.path = p
#end if
if 'files' in allowed:
self.files = set([path_string(f) for f in self.files])
#end if
if not isinstance(self.input,(self.input_type,GenericSimulationInput)):
self.error('input must be of type {0}\nreceived {1}\nplease provide input appropriate to {2}'.format(self.input_type.__name__,self.input.__class__.__name__,self.__class__.__name__))
#end if
if isinstance(self.system,PhysicalSystem):
self.system = self.system.copy()
consistent,msg = self.system.check_consistent(exit=False,message=True)
if not consistent:
locdir = os.path.join(nexus_core.local_directory,nexus_core.runs,self.path)
self.error('user provided physical system is not internally consistent\nsimulation identifier: {0}\nlocal directory: {1}\nmore details on the user error are given below\n\n{2}'.format(self.identifier,locdir,msg))
#end if
elif self.system is not None:
self.error('system must be a PhysicalSystem object\nyou provided an object of type: {0}'.format(self.system.__class__.__name__))
#end if
if self.restartable or self.force_restart:
if not cls.supports_restarts:
self.warn('restarts are not supported by {0}, request ignored'.format(cls.__name__))
#end if
#end if
#end def set
[docs]
def set_directories(self):
self.locdir = os.path.join(nexus_core.local_directory,nexus_core.runs,self.path)
self.remdir = os.path.join(nexus_core.remote_directory,nexus_core.runs,self.path)
self.resdir = os.path.join(nexus_core.local_directory,nexus_core.results,nexus_core.runs,self.path)
if not self.fake():
#print ' creating sim {0} in {1}'.format(self.simid,self.locdir)
if self.locdir not in self.sim_directories:
self.sim_directories[self.locdir] = set([self.identifier])
else:
idset = self.sim_directories[self.locdir]
if self.identifier not in idset:
idset.add(self.identifier)
else:
self.error('multiple simulations in a single directory have the same identifier\nplease assign unique identifiers to each simulation\nsimulation directory: {0}\nrepeated identifier: {1}\nother identifiers: {2}\nbetween the directory shown and the identifiers listed, it should be clear which simulations are involved\nmost likely, you described two simulations with identifier {3}'.format(self.locdir,self.identifier,sorted(idset),self.identifier))
#end if
#end if
#end if
self.image_dir = self.image_dir+'_'+self.identifier
self.imlocdir = os.path.join(self.locdir,self.image_dir)
self.imremdir = os.path.join(self.remdir,self.image_dir)
self.imresdir = os.path.join(self.resdir,self.image_dir)
#end def set_directories
[docs]
def set_files(self):
if self.infile is None:
self.infile = self.identifier + self.infile_extension
#end if
if self.outfile is None:
self.outfile = self.identifier + self.outfile_extension
#end if
if self.errfile is None:
self.errfile = self.identifier + self.errfile_extension
#end if
#end def set_files
[docs]
def reset_indicators(self):
#this is needed to support restarts
self.got_dependencies = False
self.setup = False
self.sent_files = False
self.submitted = False
self.finished = False
self.failed = False
self.got_output = False
self.analyzed = False
#end def reset_indicators
[docs]
def completed(self):
completed = self.setup
completed &= self.sent_files
completed &= self.submitted
completed &= self.finished
completed &= self.got_output
completed &= self.analyzed
completed &= not self.failed
return completed
#end def completed
[docs]
def active(self):
deps_completed = True
for dep in self.dependencies:
deps_completed &= dep.sim.completed()
#end for
active = deps_completed and not self.completed()
return active
#end def active
[docs]
def ready(self):
ready = self.active()
ready &= not self.submitted
ready &= not self.finished
ready &= not self.got_output
ready &= not self.analyzed
ready &= not self.failed
return ready
#end def ready
[docs]
def check_result(self,result_name,sim):
self.not_implemented()
#end def check_result
[docs]
def get_result(self,result_name,sim):
self.not_implemented()
#end def get_result
[docs]
def incorporate_result(self,result_name,result,sim):
self.not_implemented()
#end def incorporate_result
[docs]
def app_command(self):
self.not_implemented()
#end def app_command
[docs]
def check_sim_status(self):
self.not_implemented()
#end def check_sim_status
[docs]
def get_output_files(self): # returns list of output files to save
self.not_implemented()
#end def get_output_files
[docs]
def propagate_identifier(self):
None
#end def propagate_identifier
[docs]
def pre_init(self):
None
#end def pre_init
[docs]
def post_init(self):
None
#end def post_init
[docs]
def pre_create_directories(self):
None
#end def pre_create_directories
[docs]
def write_prep(self):
None
#end def write_prep
#end def pre_write_inputs
[docs]
def pre_send_files(self,enter):
None
#end def pre_send_files
[docs]
def post_submit(self):
None
#end def post_submit
[docs]
def pre_check_status(self):
None
#end def pre_check_status
[docs]
def post_analyze(self,analyzer):
None
#end def post_analyze
[docs]
def condense_name(self,name):
return name.strip().lower().replace('-','_').replace(' ','_')
#end def condense_name
#end def has_generic_input
[docs]
def outfile_text(self):
return self._file_text('outfile')
#end def outfile_text
[docs]
def errfile_text(self):
return self._file_text('errfile')
#end def errfile_text
def _file_text(self,filename):
filepath = os.path.join(self.locdir,self[filename])
fobj = open(filepath,'r')
text = fobj.read()
fobj.close()
return text
#end def _file_text
def _create_dir(self,dir):
if not os.path.exists(dir):
os.makedirs(dir)
elif os.path.isfile(dir):
self.error('cannot create directory {0}\na file exists at this location'.format(dir))
#end if
#end def _create_dir
[docs]
def create_directories(self):
self.pre_create_directories()
self._create_dir(self.locdir)
self._create_dir(self.imlocdir)
self.created_directories = True
#end def create_directories
[docs]
def depends(self,*dependencies):
if nexus_core.dynamic:
self.error('dynamic workflows do not allow explicit dependencies between simulations')
if len(dependencies)==0:
return
#end if
if isinstance(dependencies[0],Simulation):
dependencies = [dependencies]
#end if
for d in dependencies:
sim = d[0]
if not isinstance(sim,Simulation):
self.error('first element in a dependency tuple must be a Simulation object\nyou provided a '+sim.__class__.__name__)
#end if
dep = obj()
dep.sim = sim
rn = []
unrecognized_names = False
app_results = sim.application_results | set(['other'])
for name in d[1:]:
result_name = self.condense_name(name)
if result_name in app_results:
rn.append(result_name)
else:
unrecognized_names = True
self.error(name+' is not known to be a result of '+sim.__class__.__name__,exit=False)
#end if
#end for
if unrecognized_names:
self.error('unrecognized dependencies specified for simulation '+self.identifier)
#end if
dep.result_names = rn
dep.results = obj()
if sim.simid not in self.dependencies:
self.ordered_dependencies.append(dep)
self.dependencies[sim.simid]=dep
sim.dependents[self.simid]=self
self.dependency_ids.add(sim.simid)
self.wait_ids.add(sim.simid)
else:
self.dependencies[sim.simid].result_names.extend(dep.result_names)
#end if
#end for
#end def depends
[docs]
def undo_depends(self,sim):
i=0
for dep in self.ordered_dependencies:
if dep.sim.simid==sim.simid:
break
#end if
i+=1
#end for
self.ordered_dependencies.pop(i)
del self.dependencies[sim.simid]
del sim.dependents[self.simid]
self.dependency_ids.remove(sim.simid)
if sim.simid in self.wait_ids:
self.wait_ids.remove(sim.simid)
#end if
#end def undo_depends
# remove?
[docs]
def acquire_dependents(self,sim):
# acquire the dependents from the other simulation
dsims = obj(sim.dependents)
for dsim in dsims:
dep = dsim.dependencies[sim.simid]
dsim.depends(self,*dep.result_names)
#end for
# eliminate the other simulation
# this renders it void (fake) and removes all dependency relationships
sim.eliminate()
#end def acquire_dependents
# remove?
[docs]
def eliminate(self):
# reverse relationship of dependents (downstream)
dsims = obj(self.dependents)
for dsim in dsims:
dsim.undo_depends(self)
#end for
# reverse relationship of dependencies (upstream)
deps = obj(self.dependencies)
for dep in deps:
self.undo_depends(dep.sim)
#end for
# mark sim to be ignored in all future interactions
self.fake_sim = True
#end def eliminate
[docs]
def check_dependencies(self,result):
dep_satisfied = result.dependencies_satisfied
for dep in self.dependencies:
sim = dep.sim
for result_name in dep.result_names:
if result_name!='other':
if sim.has_generic_input():
calculating_result = False
cls = self.__class__
self.warn('a simulation result cannot be inferred from generic formatted or template input\nplease use {0} instead of {1}\nsee error below for information identifying this simulation instance'.format(cls.input_type.__class__.__name__,sim.input.__class__.__name__))
else:
calculating_result = sim.check_result(result_name,self)
#end if
if not calculating_result:
self.error('simulation {0} id {1} is not calculating result {2}\nrequired by simulation {3} id {4}\n{5} {6} directory: {7}\n{8} {9} directory: {10}'.format(sim.identifier,sim.simid,result_name,self.identifier,self.simid,sim.identifier,sim.simid,sim.locdir,self.identifier,self.simid,self.locdir),exit=False)
#end if
else:
calculating_result = True
#end if
dep_satisfied = dep_satisfied and calculating_result
#end for
#end for
result.dependencies_satisfied = dep_satisfied
#end def check_dependencies
[docs]
def get_dependencies(self):
if nexus_core.generate_only or self.finished:
for dep in self.dependencies:
for result_name in dep.result_names:
dep.results[result_name] = result_name
#end for
#end for
else:
for dep in self.dependencies:
sim = dep.sim
for result_name in dep.result_names:
if result_name!='other':
if sim.has_generic_input():
self.error('a simulation result cannot be inferred from generic formatted or template input\nplease use {0} instead of {1}\nsim id: {2}\ndirectory: {3}\nresult: {4}'.format(cls.input_type.__class__.__name__,sim.input.__class__.__name__,sim.id,sim.locdir,result_name))
#end if
dep.results[result_name] = sim.get_result(result_name,sim)
else:
dep.results['other'] = obj()
#end if
#end for
#end for
if not self.got_dependencies:
for dep in self.ordered_dependencies:
sim = dep.sim
for result_name,result in dep.results.items():
if result_name!='other':
if self.has_generic_input():
self.error('a simulation result cannot be incorporated into generic formatted or template input\nplease use {0} instead of {1}\nsim id: {2}\ndirectory: {3}\nresult: {4}'.format(cls.input_type.__class__.__name__,self.input.__class__.__name__,self.id,self.locdir,result_name))
#end if
self.incorporate_result(result_name,result,sim)
#end if
#end for
#end for
#end if
#end if
if self.renew_app_command:
self.job.renew_app_command(self)
#end if
self.got_dependencies = True
#end def get_dependencies
[docs]
def downstream_simids(self,simids=None):
if simids is None:
simids = set()
#end if
for sim in self.dependents:
simids.add(sim.simid)
sim.downstream_simids(simids)
#end for
return simids
#end def downstream_simids
[docs]
def copy_file(self,sourcefile,dest):
src = os.path.dirname(os.path.abspath(sourcefile))
dst = os.path.abspath(dest)
if src!=dst:
shutil.copy2(sourcefile,dest)
#end if
#end def copy_file
[docs]
def save_image(self,all=False):
imagefile = os.path.join(self.imlocdir,self.sim_image)
if os.path.exists(imagefile):
os.system('rm '+imagefile)
#end if
if not all:
sim_image = SimulationImage()
sim_image.save_image(self,imagefile)
else:
self.error('attempting to save full object!')
self.save(imagefile)
#end if
#end def save_image
[docs]
def load_image(self,imagepath=None,all=False):
if imagepath is None:
imagepath=os.path.join(self.imlocdir,self.sim_image)
#end if
if not all:
sim_image = SimulationImage()
sim_image.load_image(self,imagepath)
else:
self.load(imagepath)
#end if
# update process id for backwards compatibility
if 'process_id' not in self:
self.process_id = self.job.system_id
#end if
#end def load_image
[docs]
def load_analyzer_image(self,imagepath=None):
if imagepath is None:
imagepath = os.path.join(self.imresdir,self.analyzer_image)
#end if
analyzer = self.analyzer_type(self)
analyzer.load(imagepath)
return analyzer
#end def load_analyzer_image
[docs]
def save_analyzer_image(self,analyzer):
analyzer.save(os.path.join(self.imresdir,self.analyzer_image))
#end def save_analyzer_image
[docs]
def attempt_files(self):
return (self.infile,self.outfile,self.errfile)
#end def attempt_files
[docs]
def save_attempt(self):
local = self.attempt_files()
filepaths = []
for file in local:
filepath = os.path.join(self.locdir,file)
if os.path.exists(filepath):
filepaths.append(filepath)
#end if
#end for
if len(filepaths)>0:
prefix = self.identifier+'_attempt'
n=0
for dir in os.listdir(self.locdir):
if dir.startswith(prefix):
n=max(n,int(dir.replace(prefix,'')))
#end if
#end for
n+=1
attempt_dir = os.path.join(self.locdir,prefix+str(n))
os.makedirs(attempt_dir)
for filepath in filepaths:
os.system('mv {0} {1}'.format(filepath,attempt_dir))
#end for
#end if
#end def save_attempt
[docs]
def idstr(self):
return ' '+str(self.simid)+' '+str(self.identifier)
#end def idstr
#end try
#end if
#end def write_inputs
[docs]
def send_files(self,enter=True):
self.pre_send_files(enter)
if enter:
self.enter(self.locdir,False,self.simid)
#end if
self.log('sending required files'+self.idstr(),n=3)
if not os.path.exists(self.remdir):
os.makedirs(self.remdir)
#end if
if not os.path.exists(self.imremdir):
os.makedirs(self.imremdir)
#end if
if self.infile is not None:
self.files.add(self.infile)
#end if
send_files = self.files
file_locations = [self.locdir]+nexus_core.file_locations
remote = self.remdir
for file in send_files:
found_file = False
for location in file_locations:
local = os.path.join(location,file)
found_file = os.path.exists(local)
if found_file:
break
#end if
#end if
if found_file:
self.copy_file(local,remote)
else:
self.error('file {0} not found\nlocations checked: {1}'.format(file,file_locations))
#end if
#end for
self.sent_files = True
self.save_image()
send_imfiles=[self.sim_image,self.input_image]
remote = self.imremdir
for imfile in send_imfiles:
local = os.path.join(self.imlocdir,imfile)
if os.path.exists(local):
self.copy_file(local,remote)
#end if
#end for
#end def send_files
[docs]
def submit(self):
if not self.submitted:
if self.skip_submit and not self.bundled:
self.block_dependents(block_self=True)
return
#end if
self.log('submitting job'+self.idstr(),n=3)
if not self.skip_submit:
if not self.job.local:
self.job.submit()
else:
self.execute() # execute local job immediately
#end if
#end if
self.submitted = True
if (self.job.batch_mode or not nexus_core.monitor) and not nexus_core.generate_only:
self.save_image()
#end if
elif not self.finished:
self.check_status()
#end if
self.post_submit()
#end def submit
[docs]
def update_process_id(self):
if self.process_id is None and self.job.system_id is not None:
self.process_id = self.job.system_id
self.save_image()
#end if
#end def update_process_id
[docs]
def check_status(self):
self.pre_check_status()
if nexus_core.generate_only:
self.finished = self.job.finished
elif self.job.finished:
should_check = True
if self.outfile is not None:
outfile = os.path.join(self.locdir,self.outfile)
should_check &= os.path.exists(outfile)
#end if
if self.errfile is not None:
errfile = os.path.join(self.locdir,self.errfile)
should_check &= os.path.exists(errfile)
#end if
if not self.finished and should_check:
self.check_sim_status()
#end if
if self.failed:
self.finished = True
#end if
#end if
if self.finished:
self.save_image()
#end if
#end def check_status
[docs]
def get_output(self):
if not os.path.exists(self.resdir):
os.makedirs(self.resdir)
#end if
if not os.path.exists(self.imresdir):
os.makedirs(self.imresdir)
#end if
images = [self.sim_image,self.input_image]
for image in images:
remote_image = os.path.join(self.imremdir,image)
if os.path.exists(remote_image):
self.copy_file(remote_image,self.imresdir)
#end if
#end for
results_image = os.path.join(self.imresdir,self.sim_image)
if os.path.exists(results_image):
self.load_image(results_image)
#end if
if self.finished:
self.enter(self.locdir,False,self.simid)
self.log('copying results'+self.idstr(),n=3)
if not nexus_core.generate_only:
output_files = self.get_output_files()
if self.infile is not None:
output_files.append(self.infile)
#end if
if self.outfile is not None:
output_files.append(self.outfile)
#end if
if self.errfile is not None:
output_files.append(self.errfile)
#end if
files_missing = []
for file in output_files:
remfile = os.path.join(self.remdir,file)
if os.path.exists(remfile):
self.copy_file(remfile,self.resdir)
else:
files_missing.append(file)
#end if
#end for
if len(files_missing)>0:
self.log('warning: the following files were missing',n=4)
for file in files_missing:
self.log(file,n=5)
#end for
#end if
#end if
self.got_output = True
self.save_image()
#end if
#end def get_output
[docs]
def analyze(self):
if not os.path.exists(self.imresdir):
os.makedirs(self.imresdir)
#end if
if self.finished:
self.enter(self.locdir,False,self.simid)
self.log('analyzing'+self.idstr(),n=3)
if not nexus_core.generate_only:
analyzer = self.analyzer_type(self)
analyzer.analyze()
self.post_analyze(analyzer)
analyzer.save(os.path.join(self.imresdir,self.analyzer_image))
del analyzer
#end if
self.analyzed = True
self.save_image()
# support dynamic workflows
if nexus_core.dynamic:
self.fill_products()
#end if
#end def analyze
[docs]
def reset_wait_ids(self):
self.wait_ids = set(self.dependency_ids)
for sim in self.dependents:
sim.reset_wait_ids()
#end for
#end def reset_wait_ids
[docs]
def check_subcascade(self):
finished = self.finished or self.block
if not self.block and not self.block_subcascade and not self.failed:
for sim in self.dependents:
finished &= sim.check_subcascade()
#end for
#end if
self.subcascade_finished = finished
return finished
#end def check_subcascade
[docs]
def block_dependents(self,block_self=True):
if block_self:
self.block = True
#end if
self.block_subcascade = True
for sim in self.dependents:
sim.block_dependents()
#end for
#end def block_dependents
[docs]
def progress(self,dependency_id=None):
if dependency_id is not None:
self.wait_ids.remove(dependency_id)
#end if
if len(self.wait_ids)==0 and not self.block and not self.failed:
modes = nexus_core.modes
mode = nexus_core.mode
progress = True
if mode==modes.none:
return
elif mode==modes.setup:
self.write_inputs()
elif mode==modes.send_files:
self.send_files()
elif mode==modes.submit:
self.submit()
progress = self.finished
elif mode==modes.get_output:
self.get_output()
progress = self.finished
elif mode==modes.analyze:
self.analyze()
progress = self.finished
elif mode==modes.stages:
if not self.created_directories:
self.create_directories()
#end if
if not self.got_dependencies:
self.get_dependencies()
#end if
if not self.setup and 'setup' in nexus_core.stages:
self.write_inputs()
#end if
if not self.sent_files and 'send_files' in nexus_core.stages:
self.send_files()
#end if
if not self.finished and 'submit' in nexus_core.stages:
self.submit()
#end if
if nexus_core.dependent_modes <= nexus_core.stages_set:
progress_post = self.finished
progress = self.finished and self.analyzed
else:
progress_post = progress
#end if
if progress_post:
if not self.got_output and 'get_output' in nexus_core.stages:
self.get_output()
#end if
if not self.analyzed and 'analyze' in nexus_core.stages:
self.analyze()
#end if
#end if
elif mode==modes.all:
if not self.setup:
self.write_inputs()
self.send_files(False)
#end if
if not self.finished:
self.submit()
#end if
if self.finished:
if not self.got_output:
self.get_output()
#end if
if not self.analyzed:
self.analyze()
#end if
#end if
progress = self.finished
#end if
if progress and not self.block_subcascade and not self.failed:
for sim in self.dependents:
if not sim.bundled:
sim.progress(self.simid)
#end if
#end for
#end if
elif len(self.wait_ids)==0 and self.force_write:
modes = nexus_core.modes
mode = nexus_core.mode
if mode==modes.stages:
if not self.got_dependencies:
self.get_dependencies()
#end if
if 'setup' in nexus_core.stages:
self.write_inputs()
#end if
if not self.sent_files and 'send_files' in nexus_core.stages:
self.send_files()
#end if
#end if
#end if
#end def progress
[docs]
def reconstruct_cascade(self):
imagefile = os.path.join(self.imlocdir,self.sim_image)
if os.path.exists(imagefile) and not self.loaded:
self.load_image()
# continue from interruption
if self.submitted and not self.finished and self.process_id is not None:
if nexus_core.dynamic:
machine = get_machine(Job.machine)
if isinstance(machine,Workstation):
# fully rerun following interrupt
self.save_attempt()
self.reset_indicators()
self.job.system_id = self.process_id # load process id of job
self.job.reenter_queue()
#end if
self.loaded = True
#end if
for sim in self.dependents:
sim.reconstruct_cascade()
#end for
return self
#end def reconstruct_cascade
[docs]
def traverse_cascade(self,operation,*args,**kwargs):
if 'dependency_id' in kwargs:
self.wait_ids.remove(kwargs['dependency_id'])
del kwargs['dependency_id']
#end if
if len(self.wait_ids)==0:
operation(self,*args,**kwargs)
for sim in self.dependents:
kwargs['dependency_id'] = self.simid
sim.traverse_cascade(operation,*args,**kwargs)
#end for
#end if
#end def traverse_cascade
# used only in tests
[docs]
def traverse_full_cascade(self,operation,*args,**kwargs):
operation(self,*args,**kwargs)
for sim in self.dependents:
sim.traverse_full_cascade(operation,*args,**kwargs)
#end for
#end def traverse_full_cascade
[docs]
def write_dependents(self,n=0,location=False,block_status=False):
outs = [self.__class__.__name__,self.identifier,self.simid]
if location:
outs.append(self.locdir)
#end if
if block_status:
if self.block:
outs.append('blocked')
else:
outs.append('unblocked')
#end if
#end if
outs.append(list(self.dependency_ids))
self.log(*outs,n=n)
n+=1
for sim in self.dependents:
sim.write_dependents(n=n,location=location,block_status=block_status)
#end for
#end def write_dependents
[docs]
def execute(self,run_command=None):
pad = self.enter(self.locdir)
if run_command is None:
job = self.job
command = 'export OMP_NUM_THREADS='+str(job.threads)+'\n'
if len(job.presub)>0:
command += job.presub+'\n'
#end if
machine = job.get_machine()
command += job.run_command(machine.app_launcher)
if len(job.postsub)>0:
command += job.postsub+'\n'
#end if
command = ('\n'+command).replace('\n','\n '+pad)
run_command = command
#end if
if self.job is None or self.job.env is None:
env = os.environ.copy()
else:
env = job.env
#end if
if nexus_core.generate_only:
self.log(pad+'Would have executed: '+command)
else:
self.log(pad+'Executing: '+command)
fout = open(self.outfile,'w')
ferr = open(self.errfile,'w')
out,err = Popen(command,env=env,stdout=fout,stderr=ferr,shell=True,close_fds=True).communicate()
#end if
self.leave()
self.submitted = True
if self.job is not None:
job.status = job.states.finished
self.job.finished = True
#end if
#end def execute
#end if
#end def show_input
# dynamic workflow support
[docs]
def fill_produces(self):
self.not_implemented('fill_produces')
#end def fill_produces
[docs]
def fill_products(self):
self.not_implemented('fill_products')
#end def fill_products
#end class Simulation
#end def return_system
#end class NullSimulationInput
[docs]
class NullSimulationAnalyzer(SimulationAnalyzer):
def __init__(self,sim):
None
#end def __init__
[docs]
def analyze(self):
None
#end def analyze
#end class NullSimulationAnalyzer
#end class GenericSimulationInput
[docs]
class GenericSimulation(Simulation):
allowed_inputs = Simulation.allowed_inputs | set(['outfiles'])
def __init__(self,**kwargs):
import os
self.outfiles = kwargs.pop('outfiles',[])
if 'input' in kwargs:
input = kwargs['input']
if isinstance(input,str):
if os.path.exists(input):
self.infile = input
kwargs['input'] = input_template(filepath=self.infile)
else:
text = input
kwargs['input'] = input_template(text=text)
#end if
#end if
self.input_type = NullSimulationInput
self.analyzer_type = NullSimulationAnalyzer
if 'input_type' in kwargs:
self.input_type = kwargs['input_type']
del kwargs['input_type']
#end if
if 'analyzer_type' in kwargs:
self.analyzer_type = kwargs['analyzer_type']
del kwargs['analyzer_type']
#end if
if 'input' in kwargs:
self.input_type = kwargs['input'].__class__
#end if
if 'analyzer' in kwargs:
self.analyzer_type = kwargs['analyzer'].__class__
#end if
Simulation.__init__(self,**kwargs)
#end def __init__
[docs]
def check_sim_status(self):
import os
outfiles = self.get_output_files()
files_exist = True
for f in outfiles:
fp = os.path.join(self.locdir,f)
fp_exists = os.path.exists(fp)
files_exist &= fp_exists
#end for
self.failed = not files_exist
self.finished = not self.failed
#end def check_sim_status
[docs]
def get_output_files(self):
return self.outfiles
#end def get_output_files
[docs]
def app_command(self):
if self.job.app_name is not None:
return self.job.app_name+' '+self.infile
else:
return self.job.app_command
#end if
#end def app_command
#end class GenericSimulation
#end def preprocess
#end class SimulationInputTemplateDev
#end for
#end if
#end def write
#end class SimulationInputMultiTemplateDev
# these are for user access, *Dev are for development
#end class SimulationInputTemplate
#end class SimulationInputMultiTemplate
# developer functions
#end def input_template_dev
#end def multi_input_template_dev
# user functions
#end def input_template
#end def multi_input_template
#end def generate_template_input
#end def generate_multi_template_input
[docs]
def generate_simulation(**kwargs):
sim_type='generic'
if 'sim_type' in kwargs:
sim_type = kwargs['sim_type']
del kwargs['sim_type']
#end if
if sim_type=='generic':
return GenericSimulation(**kwargs)
else:
Simulation.class_error('sim_type {0} is unrecognized'.format(sim_type),'generate_simulation')
#end if
#end def generate_simulation
# ability to graph simulation workflows
try:
from pydot import Dot,Node,Edge
except:
Dot,Node,Edge = unavailable('pydot','Dot','Node','Edge')
#end try
try:
from matplotlib.image import imread
from matplotlib.pyplot import imshow,show,xticks,yticks
except:
imread = unavailable('matplotlib.image','imread')
imshow,show,xticks,yticks = unavailable('matplotlib.pyplot','imshow','show','xticks','yticks')
#end try
exit_call = sys.exit
[docs]
def graph_sims(sims=None,savefile=None,useid=False,exit=True,quants=True,display=True):
if sims is None:
sims = Simulation.all_sims
#end if
graph = Dot(graph_type='digraph',dpi=300)
graph.set_label('simulation workflows')
graph.set_labelloc('t')
nodes = obj()
for sim in sims:
if 'fake_sim' in sim and sim.fake_sim:
continue
#end if
if sim.simlabel is not None and not useid:
nlabel = sim.simlabel+' '+str(sim.simid)
else:
nlabel = sim.identifier+' '+str(sim.simid)
#end if
nopts = obj()
if 'block' in sim and sim.block:
nopts.color = 'black'
nopts.fontcolor = 'white'
#end if
node = obj(
id = sim.simid,
sim = sim,
node = Node(nlabel,style='filled',shape='Mrecord',**nopts),
edges = obj(),
)
nodes[node.id] = node
graph.add_node(node.node)
#end for
for node in nodes:
for simid,dep in node.sim.dependencies.items():
other = nodes[simid].node
if quants:
for quantity in dep.result_names:
edge = Edge(other,node.node,label=quantity,fontsize='10.0')
graph.add_edge(edge)
#end for
else:
edge = Edge(other,node.node)
graph.add_edge(edge)
#end if
#end for
#end for
if savefile is None:
fout = tempfile.NamedTemporaryFile(suffix='.png')
savefile = fout.name
#savefile = './sims.png'
#end if
fmt = savefile.rsplit('.',1)[1]
graph.write(savefile,format=fmt,prog='dot')
# display the image
if fmt=='png' and display:
imshow(imread(savefile))
xticks([])
yticks([])
show()
#end if
if exit:
exit_call()
#end if
#end def graph_sims
[docs]
class DynamicProcess(DevBase):
'''Enables dynamic workflows execution
Basic DP contains a single simulation.
Derived classes may perform more elaborate processes,
i.e. recovery for failed jobs, resetting the primary
simulation object (sim data member) to point at the
final sim in the process.
Takes the place of Simulation in user scripts. All
generate_* simulation functions return DP's when
executing dynamic workflows.
'''
all_dynamic_processes = obj()
allowed_requirements = set([
'none',
'structure',
'charge_density',
'orbitals',
'jastrow',
'wavefunction',
'pwscf_orbitals', # explicit QE
])
[docs]
@classmethod
def check_first_gen(cls,kw):
nc_loc = nexus_core.local_directory
runs = nexus_core.runs
path = kw['path']
identifier = kw['identifier']
locdir = os.path.join(nc_loc,runs,path)
if 'dynamic_id' not in kw:
cls.class_error('dynamic_id is required for dynamic workflows in a generate_* function.\nSimulation run location: {}\nSimulation identifier : {}'.format(locdir,identifier))
dynamic_id = kw.pop('dynamic_id')
dpid = (locdir,identifier,dynamic_id)
if dpid in DynamicProcess.all_dynamic_processes:
dp = DynamicProcess.all_dynamic_processes[dpid]
return dp,None
else:
dp = None
if 'requires' not in kw:
cls.class_error('dependency requirements must be given via the "requires" keyword for dynamic workflows')
requires = kw.pop('requires')
dyn_args = obj(dpid=dpid,requires=requires)
return dp,dyn_args
#end def check_first_gen
def __init__(self,dpid,sim,requires):
# check dynamic id
if dpid in self.all_dynamic_processes:
self.error('dynamic process created with overlapping id. Provided id: {}'.format(dpid))
# check simulation type
if not isinstance(sim,Simulation):
self.error('expected Simulation type but received type {}'.format(sim.__class__.__name__))
# check requires
if isinstance(requires,str):
requires = [requires]
elif not isinstance(requires,(tuple,list,set)):
self.error('keyword "requires" must be a tuple, list or set of requirements')
for req in requires:
if not isinstance(req,str):
self.error('each requirement in "requires" must be given as a string.\nType received: {}\nValue received: {}'.format(req.__class__.__name__,req))
requires = set(requires)
invalid_reqs = requires-self.allowed_requirements
if len(invalid_reqs)>0:
self.error('invalid requirements provided.\nAllowed requirements: {}\nRequirements provided: {}'.format(list(self.allowed_requirements),list(invalid_reqs)))
if len(requires)==0:
self.error("every simulation dynamic process must specify least one dependency requirement.\nIf there are no dependencies/requirements, set requires='none'")
if 'none' in requires:
requires.remove('none')
# check produces
produces = sim.produces
if isinstance(produces,str):
produces = [produces]
if not isinstance(produces,(tuple,list,set)):
self.error('keyword "requires" must be a tuple, list or set of products')
for prod in produces:
if not isinstance(req,str):
self.error('each product in "produces" must be given as a string.\nType received: {}\nValue received: {}'.format(req.__class__.__name__,req))
produces = set(produces)
# initial values
self.dpid = dpid # unique identifier, str
self.sim = sim # wrapped Simulation object
self.requires = requires # replaces dependencies
self.unmet_reqs = set(requires)
self.req_values = obj()
self.reqs_met = False
self.produces = produces
# store references in global registries
self.all_dynamic_processes[dpid] = self
dynamic_storage.dynamic_processes[dpid] = self
dynamic_storage.dynamic_process_ids.add(dpid)
#end def __init__
[docs]
def requirements_met(self):
'''Check if all input/dependency requirements are met'''
if self.reqs_met:
return True
reqs_met = True
reqs_met &= len(self.unmet_reqs)==0
reqs_met &= len(self.requires-set(self.req_values))==0
if reqs_met:
self.reqs_met = reqs_met
return reqs_met
#end def requirements_met
def _check_get_product(self,prod_name):
'''Support product getter functions
Note that requirements are a subset of products
'''
sim = self.sim
msg = None
if prod_name not in sim.produces:
msg = 'simulation does not produce "{}"'.format(prod_name)
elif not sim.finished:
msg = 'Simulation is not finished\nProduct "{}" not yet computed'.format(prod_name)
elif not sim.analyzed:
msg = 'simulation has not been analyzed, requested prod_name "{}" has not been computed yet'.format(prod_name)
elif prod_name not in sim.products:
msg = 'simulation products have not been handled correctly. This is a developer error'
if msg is not None:
self.error(msg+'\nSimulation type : {}\nSimulation id : {}\nSimulation directory: {}\nDynamic process id : {}'.format(sim.__class__.__name__,sim.simid,sim.locdir,self.dpid))
return sim.products[prod_name]
#end def _check_get_product
def _check_set_requirement(self,
req_name,
req_value = None,
req_type = str,
is_path = False,
):
'''Support requirement setter functions'''
# check supported requirement value types
if not isinstance(req_value,req_type):
if not isinstance(req_type,tuple):
ts = req_type.__name__
else:
ts = [t.__name__ for t in req_type]
self.error('product "{}" must be of type "{}".\nReceived type: {}'.format(req_name,ts,req_value.__class__.__name__))
# check if requirement value has already been set
if req_name not in self.req_values:
self.req_values[req_name] = req_value
already_set = False
elif isinstance(req_value,(str,int)) and req_value!=self.req_values[req_name]:
self.error('attempted assignment of required parameter "{}" with value differing from the original.\nOriginal value: {}\nValue received: {}'.format(req_name,self.req_values[req_name],req_value))
elif id(req_value)!=id(self.req_values[req_name]):
self.error('attempted assignment of required parameter "{}" with python id differing from the original.\nOriginal id: {}\nid received: {}'.format(req_name,id(self.req_values[req_name]),id(req_value)))
else:
already_set = True
# if already set, return
if already_set:
return already_set
# proceed with incorporation
# ensure requirement is one of the supported options in general
if req_name not in self.sim.allowed_requirements:
self.error('incorporating "{}" into simulation type {} is not supported.'.format(req_name,self.sim.__class__.__name__))
elif is_path and isinstance(req_value,str) and not os.path.exists(req_value):
self.error('"{}" path does not exist.\nPath provided: {}'.format(req_name,req_value))
# mark the requirement as fulfilled
self.unmet_reqs.remove(req_name)
return already_set
#end def _check_set_requirement
# general access to product info
#@property
#def produces(self):
# return self.sim.produces
@property
def products(self):
return self.sim.products
# getters for all possible requirements (subset of products)
@property
def structure(self):
return self._check_get_product('structure')
@property
def charge_density(self):
return self._check_get_product('charge_density')
@property
def orbitals(self):
return self._check_get_product('orbitals')
@property
def jastrow(self):
return self._check_get_product('jastrow')
@property
def wavefunction(self):
return self._check_get_product('wavefunction')
@property
def pwscf_orbitals(self):
return self._check_get_product('pwscf_orbitals')
# setters for all possible requirements
@structure.setter
def structure(self,struct):
already_set = self._check_set_requirement(
'structure',struct,req_type=(str,Structure),is_path=True)
if already_set:
return
if isinstance(struct,str):
struct = read_structure(struct)
else:
struct = struct.copy()
self.sim.receive_structure(struct)
#end def structure
@charge_density.setter
def charge_density(self,charge_density):
already_set = self._check_set_requirement(
'charge_density',charge_density,is_path=True)
if already_set:
return
self.sim.receive_charge_density(charge_density)
#end def charge_density
@orbitals.setter
def orbitals(self,orbitals):
already_set = self._check_set_requirement(
'orbitals',orbitals,is_path=True)
if already_set:
return
self.sim.receive_orbitals(orbitals)
#end def orbitals
@jastrow.setter
def jastrow(self,jastrow):
already_set = self._check_set_requirement(
'jastrow',jastrow,is_path=True)
if already_set:
return
self.sim.receive_jastrow(jastrow)
#end def jastrow
@wavefunction.setter
def wavefunction(self,wavefunction):
already_set = self._check_set_requirement(
'wavefunction',wavefunction,is_path=True)
if already_set:
return
self.sim.receive_wavefunction(wavefunction)
#end def wavefunction
@pwscf_orbitals.setter
def pwscf_orbitals(self,pwscf_orbitals):
already_set = self._check_set_requirement(
'pwscf_orbitals',pwscf_orbitals,is_path=True)
if already_set:
return
self.sim.receive_pwscf_orbitals(pwscf_orbitals)
#end def pwscf_orbitals
# preserve Simulation UI
# data fields and functions
@property
def simid(self):
return self.sim.simid
@property
def identifier(self):
return self.sim.identifier
@property
def job(self):
return self.sim.job
@property
def input(self):
return self.sim.input
@input.setter
def input(self,input):
self.sim.input = input
@property
def system(self):
return self.sim.system
@property
def analyzer_image(self):
return self.sim.analyzer_image
# status_flags
@property
def setup(self):
return self.sim.setup
@property
def sent_files(self):
return self.sim.sent_files
@property
def submitted(self):
return self.sim.submitted
@property
def finished(self):
return self.sim.finished
@property
def got_output(self):
return self.sim.got_output
@property
def analyzed(self):
return self.sim.analyzed
@property
def failed(self):
return self.sim.failed
# execution modification
@property
def skip_submit(self):
return self.sim.skip_submit
@property
def block(self):
return self.sim.block
# try on new user-facing status properties
@property
def done(self):
return self.sim.finished
@property
def succ(self):
return self.sim.finished and not self.sim.failed
@property
def fail(self):
return self.sim.failed
#return self.sim.finished and self.sim.failed
#end class DynamicProcess
