''' Automatic verification that the ANUGA code runs the Patong simulation and produces the expected output. Required files are downloaded from the ANUGA servers if they are out of date or missing. ''' import os import glob import unittest import time import shutil import project from anuga.utilities.system_tools import get_web_file, untar_file import anuga.utilities.log as log # base URL for the remote ANUGA data PATONG_DATA_URL = 'http://10.7.64.243/patong_validation_data/' # path to the local data directory Local_Data_Directory = os.path.join('.', 'local_data') # path to the remote data directory Remote_Data_Directory = os.path.join('.', 'remote_data') # sequence of required local data objects # first is always the SWW file to be compared Local_Data_Objects = ('patong.sww', 'data') # name of stdout catch file for runmodel.py RUNMODEL_STDOUT = 'runmodel.stdout' # text at start of 'output dir' line in RUNMODEL_STDOUT file OUTDIR_PREFIX = 'Make directory ' # Name of SWW file produced by simulation OUTPUT_SWW = 'patong.sww' def setup(): '''Prepare for the validation run. Check we have required data set in project.py. ''' # Check that environment variables are defined. if os.getenv(project.ENV_INUNDATIONHOME) is None: msg = ("Environment variable '%s' is not set in project.py" % project.ENV_INUNDATIONHOME) raise Exception, msg if os.getenv(project.ENV_MUXHOME) is None: msg = ("Environment variable '%s' is not set in project.py" % project.ENV_MUXHOME) raise Exception, msg def update_local_data(): '''Update local data objects from the server. These are expected to be *.tgz files/directories. ''' # local function to update one data object def update_object(obj, auth): '''Update object 'obj' using authentication tuple 'auth'.''' # Get a unique date+time string to defeat caching. The idea is to add # this to the end of any URL so proxy sees a different request. cache_defeat = '?' + time.strftime('%Y%m%d%H%M%S') # create local and remote paths, URLs, etc. remote_path = os.path.join(Remote_Data_Directory, obj) remote_digest = remote_path + '.tgz.digest' local_path = os.path.join(Local_Data_Directory, obj) local_file = local_path + '.tgz' local_digest = local_file + '.digest' object_url = PATONG_DATA_URL + obj + '.tgz' digest_url = object_url + '.digest' # see if missing either digest or object .tgz if not os.path.exists(local_digest) or not os.path.exists(local_file): # no digest or no object, download both digest and object log.debug('Fetching remote file %s' % digest_url) auth = get_web_file(digest_url+cache_defeat, local_digest, auth=auth) log.debug('Fetching remote file %s' % object_url) auth = get_web_file(object_url+cache_defeat, local_file, auth=auth) else: # download object digest to remote data directory log.debug('Fetching remote file %s' % object_url) auth = get_web_file(digest_url+cache_defeat, remote_digest, auth=auth) # compare remote with local digest fd = open(local_digest, 'r') local_data_digest = fd.read() fd.close() fd = open(remote_digest, 'r') remote_data_digest = fd.read() fd.close() # if digests differ, refresh object if local_data_digest != remote_data_digest: log.debug('Local file %s is out of date' % obj) fd = open(local_digest, 'w') fd.write(remote_data_digest) fd.close() log.debug('Fetching remote file %s' % object_url) auth = get_web_file(object_url+cache_defeat, local_file, auth=auth) return auth log.debug('Refreshing local data ...') # create local data directory if required if not os.path.exists(Local_Data_Directory): os.mkdir(Local_Data_Directory) # clean out remote data copy directory shutil.rmtree(Remote_Data_Directory, ignore_errors=True) os.mkdir(Remote_Data_Directory) # refresh local files auth = None for data_object in Local_Data_Objects: auth = update_object(data_object, auth) # unpack *.tgz files for data_object in Local_Data_Objects: tar_path = os.path.join(Local_Data_Directory, data_object+'.tgz') log.debug('Untarring %s in dir %s' % (tar_path, Local_Data_Directory)) untar_file(tar_path, target_dir=Local_Data_Directory) log.debug('Local data has been refreshed.') def run_simulation(): '''Run the Patong simulation.''' # modify environment so we use the local data # INUNDATIONHOME points into the 'data' local data directory old_inundationhome = os.getenv(project.ENV_INUNDATIONHOME) os.putenv(project.ENV_INUNDATIONHOME, os.path.join(Local_Data_Directory, '')) old_muxhome = os.getenv(project.ENV_MUXHOME) os.putenv(project.ENV_MUXHOME, os.path.join(Local_Data_Directory, 'data')) # run the simulation, produce SWW file log.debug('Running Patong simulation ...') cmd = 'python run_model.py > %s' % RUNMODEL_STDOUT res = os.system(cmd) assert res == 0 # undo environment changes if old_inundationhome: os.putenv(project.ENV_INUNDATIONHOME, old_inundationhome) if old_muxhome: os.putenv(project.ENV_MUXHOME, old_muxhome) def check_that_output_is_as_expected(): '''Check that validation output is as required.''' # get path to expected SWW file log.debug('Checking that simulation results are as expected ...') local_sww = os.path.join(Local_Data_Directory, Local_Data_Objects[0]) # get output directory from stdout capture file try: fd = open(RUNMODEL_STDOUT, 'r') except IOError, e: log.critical("Can't open catch file '%s': %s" % (RUNMODEL_STDOUT, str(e))) return 1 lines = fd.readlines() fd.close output_directory = None for line in lines: if line.startswith(OUTDIR_PREFIX): output_directory = line.replace(OUTDIR_PREFIX, '', 1) output_directory = output_directory.strip('\n') break if output_directory is None: log.critical("Couldn't find line starting with '%s' in file '%s'" % (OUTDIR_PREFIX, RUNMODEL_STDOUT)) return 1 # compare SWW files here and there new_output_sww = os.path.join(output_directory, OUTPUT_SWW) cmd = 'python cmpsww.py %s %s > cmpsww.stdout' % (local_sww, new_output_sww) res = os.system(cmd) assert res == 0 log.debug('Simulation results are as expected.') def teardown(): '''Clean up after validation run.''' # clear all data objects from local data directory for data_object in Local_Data_Objects: obj_path = os.path.join(Local_Data_Directory, data_object) if os.path.isfile(obj_path): os.remove(obj_path) else: shutil.rmtree(obj_path, ignore_errors=True) # remove remote directory and stdout capture file shutil.rmtree(Remote_Data_Directory, ignore_errors=True) os.remove(RUNMODEL_STDOUT) ################################################################################ # Mainline - run the simulation, check output. ################################################################################ # set logging levels log.console_logging_level = log.DEBUG setup() # make sure local data is up to date update_local_data() # run the simulation run_simulation() # check output is as expected check_that_output_is_as_expected() # clean up teardown()