CoreFlows/swig/Validation/cf_resultBase_loader.py

   1 import resultsBase as rb
   2 import re
   3 import numpy as np
   4 import VTU_routines
   5 import MED_routines
   6 import CoreFlows_variables_dictionary
   7
   8 solver_dict_cf = {
   9     'cf-ordre1': 'CF-ordre1',
  10     'vfroe_': 'CF-ordre1',
  11     'vfroecorp': 'CF-ordre2',
  12     'cf-ordre2': 'CF-ordre2',
  13     'vffc_wb': 'VFFCeq',
  14     'pstag_wb': 'pSTAGeq',
  15     'ps_wb': 'pSTAGeq',
  16     'staggered_wb': 'pSTAGeq'
  17 }
  18
  19 def create_file_list(workDir, ext=None, fname_glob=None):
  20     expanded_path = os.path.expanduser(workDir)
  21     abs_path = os.path.abspath(expanded_path)
  22     # TODO : we should test if directory exists
  23     if fname_glob is None:
  24         pattern_for_files = abs_path + '/*' + ext
  25     elif ext is None:
  26         pattern_for_files = abs_path + '/' + fname_glob
  27     file_list = glob.glob(pattern_for_files)
  28     return file_list
  29
  30 # TODO : we should use regexp instead?
  31
  32
  33 def get_variable_name_from_string(str, varDict, default_value="noVar"):
  34     res = default_value
  35     for var_name in varDict.keys():
  36         if re.search(var_name, str,re.IGNORECASE):
  37             res = varDict[var_name]
  38             break
  39     return res
  40
  41
  42 # deprecated: use the last_step property instead
  43 def get_last_step(resDb):
  44     res = max([elt.step for elt in resDb])
  45     return res
  46
  47 # deprecated: use last_step_elts property instead
  48 def extract_last_step_elt(resDb):
  49     last_step = get_last_step(resDb)
  50     last_step_elts = [elt for elt in resDb if elt.step == last_step]
  51     if len([elt for elt in last_step_elts if elt.is_stationnary == True]) > 0:
  52         logging.warning(
  53             "you are extracting last step variables "
  54             "from a result list that contains stationnary results.\n"
  55         )
  56     return last_step_elts
  57
  58
  59 def is_stationnary_cf(name):
  60     pattern_for_stationnary = ".*_Stat_.*"
  61     match = re.search(pattern_for_stationnary, name)
  62     return (match != None)
  63
  64
  65 def create_result_list_1D(dir_list, tags=[], fname_glob=None, code='cf', solver=None, search_step=True):
  66     if len(dir_list) == 0:
  67         logging.warning(
  68             "You passed an empty list of dir to CreateResultList_CoreFlows")
  69         return []
  70     else:
  71         result_list = []
  72         for a_dir in dir_list:
  73             if fname_glob == None:
  74                 fileList = rb.create_file_list(a_dir, ext='csv')
  75             else:
  76                 fileList = rb.create_file_list(a_dir, fname_glob=fname_glob)
  77             for f in fileList:
  78                 variable = rb.get_variable_name_from_string(
  79                     f, var_dict_cf, default_value='unknown_variable')
  80                 if search_step :
  81                     step = rb.get_step_from_root_name(f)
  82                 else:
  83                     step = None
  84                 is_stationnary = is_stationnary_cf(f)
  85                 if solver == None:
  86                     result_solver = rb.get_solver_name(f, solver_dict_cf)
  87                 else:
  88                     result_solver = solver
  89                 field_result = rb.Result(
  90                     file_name=f,
  91                     variable=variable,
  92                     dimension=1,
  93                     step=step,
  94                     code=code,
  95                     is_stationnary=is_stationnary,
  96                     solver=result_solver,
  97                     tags=tags
  98                 )
  99                 result_list.append(field_result)
 100     return result_list
 101
 102
 103 def create_result_list_2D(dir_list, tags=[], solver=None):
 104     if len(dir_list) == 0:
 105         logging.warning(
 106             "You passed an empty list of dir to CreateResultList_CoreFlows")
 107         return []
 108     else:
 109         result_list = []
 110         for a_dir in dir_list:
 111             files = rb.create_file_list(a_dir, ext='vtu')
 112             for f in files:
 113                 variable = rb.get_variable_name_from_string(
 114                     f, var_dict_cf_2D, default_value='unknown_variable')
 115                 step = rb.get_step_from_root_name(f)
 116                 is_stationnary = is_stationnary_cf(f)
 117                 if solver == None:
 118                     result_solver = rb.get_solver_name(f, solver_dict_cf)
 119                 else:
 120                     result_solver = solver
 121                 field_result = rb.Result(
 122                     file_name=f,
 123                     variable=variable,
 124                     dimension=2,
 125                     step=step,
 126                     code='cf',
 127                     is_stationnary=is_stationnary,
 128                     solver=result_solver,
 129                     tags=tags
 130                 )
 131                 result_list.append(field_result)
 132     return result_list
 133
 134 # for the sake of compatibility with ancient version
 135 CreateResultList_2D = create_result_list_2D